FastAPI+SQLAlchemy多对多关系报错：'NoneType' object has no attribute 'append'

在尝试通过FastAPI和SQLAlchemy构建多对多关系时，执行用户添加到公司的操作时出错，错误出现在await user.users_companies.append(company)行。

相关代码 #

模型定义

companies_users = Table(
    'companies_users',
    Base.metadata,
    Column('user_id', ForeignKey('users.id')),
    Column('company_id', ForeignKey('companies.id')),
    Column('is_active_owner', Boolean),
    Column('is_active_user', Boolean)
)

class DbUser(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=False)
    email = Column(String, nullable=False, unique=True)
    password = Column(String, nullable=False)
    bio = Column(String, default=None, nullable=True)
    time_created = Column(DateTime(timezone=True), server_default=func.now())
    time_updated = Column(DateTime(timezone=True), onupdate=func.now())
    users_companies = relationship(
        'DbCompany', secondary=companies_users, backref='companies_users'
    )
    # 注：此处companies_administrators未定义，需确保已正确声明
    company_administrators = relationship('DbCompany', secondary=companies_administrators)

users = DbUser.__table__

class DbCompany(Base):
    __tablename__ = 'companies'
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=False)
    description = Column(String, nullable=False)
    is_hide = Column(Boolean, nullable=False)
    owner = Column(String, ForeignKey('users.email'))

companies = DbCompany.__table__

业务函数

async def add_user_to_company_a(self, company_id: int, user_id: int, current_user: UserDisplayWithId):
    user = await self.database.fetch_one(users.select().where(users.c.id == user_id))
    company = await self.database.fetch_one(companies.select().where(companies.c.id == company_id))
    if user is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f'User with id {user_id} not found')
    elif company is None:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f'Company with id {company_id} not found')
    else:
        await user.users_companies.append(company)
        raise HTTPException(status_code=status.HTTP_201_CREATED,
                            detail=f'User {user_id} successfully added to company {company_id}')

问题原因 #

1. 返回对象类型错误：self.database.fetch_one()返回的是数据库原生的行对象（Row），不是SQLAlchemy的DbUser/DbCompany模型实例，因此不存在users_companies这个ORM关系属性。
2. 不必要的await：append()是Python列表的同步方法，不需要用await修饰，这也是语法错误的来源之一。
3. 关系定义潜在问题：DbUser中的company_administrators依赖未定义的companies_administrators表，可能导致ORM初始化异常。

解决方法 #

方法一：使用SQLAlchemy异步会话操作模型实例

SQLAlchemy异步模式下，应使用AsyncSession来查询和操作模型实例，而非直接操作表对象。

1. 先修正模型的关系定义（推荐用back_populates替代backref，逻辑更清晰）：

class DbUser(Base):
    __tablename__ = 'users'
    # ... 其他字段省略
    users_companies = relationship(
        'DbCompany', secondary=companies_users, 
        back_populates='company_users'
    )

class DbCompany(Base):
    __tablename__ = 'companies'
    # ... 其他字段省略
    company_users = relationship(
        'DbUser', secondary=companies_users,
        back_populates='users_companies'
    )

2. 修改业务函数，使用AsyncSession：

from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select

async def add_user_to_company_a(self, db: AsyncSession, company_id: int, user_id: int, current_user: UserDisplayWithId):
    # 查询模型实例
    user_result = await db.execute(select(DbUser).where(DbUser.id == user_id))
    user = user_result.scalar_one_or_none()
    
    company_result = await db.execute(select(DbCompany).where(DbCompany.id == company_id))
    company = company_result.scalar_one_or_none()

    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f'User with id {user_id} not found')
    elif not company:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f'Company with id {company_id} not found')
    
    # 操作关系，无需await
    user.users_companies.append(company)
    # 若需设置中间表的额外字段，可通过关联对象的attributes操作：
    # user.users_companies[-1].is_active_user = True
    await db.commit()
    await db.refresh(user)  # 可选，刷新实例以获取最新关系数据
    
    raise HTTPException(status_code=status.HTTP_201_CREATED,
                        detail=f'User {user_id} successfully added to company {company_id}')

方法二：直接操作中间表插入数据

因为中间表包含额外字段（is_active_owner、is_active_user），直接插入更灵活，无需依赖ORM关系：

async def add_user_to_company_a(self, company_id: int, user_id: int, current_user: UserDisplayWithId):
    # 先校验用户和公司存在
    user = await self.database.fetch_one(users.select().where(users.c.id == user_id))
    company = await self.database.fetch_one(companies.select().where(companies.c.id == company_id))

    if not user:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f'User with id {user_id} not found')
    elif not company:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f'Company with id {company_id} not found')
    
    # 插入中间表数据，根据业务需求设置额外字段
    await self.database.execute(
        companies_users.insert().values(
            user_id=user_id,
            company_id=company_id,
            is_active_owner=False,
            is_active_user=True
        )
    )
    
    raise HTTPException(status_code=status.HTTP_201_CREATED,
                        detail=f'User {user_id} successfully added to company {company_id}')

内容的提问来源于stack exchange，提问作者Oleksandr Kovmir