FastAPI异步模式下Item创建触发IntegrityError及用户读取问题

报错信息 #

raise translated_error from error
sqlalchemy.exc.IntegrityError: (sqlalchemy.dialects.postgresql.asyncpg.IntegrityError) <class 'asyncpg.exceptions.NotNullViolationError'>: null value in column "owner_id" of relation "items" violates not-null constraint
DETAIL:  Failing row contains (9, string, string, null).
[SQL: INSERT INTO items (title, description, owner_id) VALUES ($1::VARCHAR, $2::VARCHAR, $3::INTEGER) RETURNING items.id]
[parameters: ('string', 'string', None)]

问题分析 #

创建Item时触发NotNullViolationError，尽管代码中手动传入了owner_id，但实际插入时该值为None；同时存在无法读取已创建User的问题，核心原因是异步SQLAlchemy模式下，模型关联的懒加载配置错误、会话操作未同步数据库数据。

修复方案 #

1. 修正模型关联的加载配置 #

异步SQLAlchemy不支持同步模式的lazy参数，需移除该配置，关联数据的加载改为查询时显式指定策略。修改models.py：

# models.py 修改后
import sqlalchemy as sa
import sqlalchemy.orm as so
from database import Base

class Item(Base):
    __tablename__ = "items"

    id: so.Mapped[int] = so.mapped_column(sa.Integer, primary_key=True, index=True)
    title: so.Mapped[str] = so.mapped_column(sa.String, index=True)
    description: so.Mapped[str] = so.mapped_column(sa.Text, index=True)
    owner_id: so.Mapped[int] = so.mapped_column(
        sa.ForeignKey('users.id'), index=True)

    user: so.Mapped['User'] = so.relationship(
        back_populates='items'
    )

    def __repr__(self):
        return f'Item({self.id} "{self.title}")'

class User(Base):
    __tablename__ = "users"

    id: so.Mapped[int] = so.mapped_column(sa.Integer, primary_key=True, index=True)
    username: so.Mapped[str] = so.mapped_column(sa.String, unique=True, index=True)
    email: so.Mapped[str] = so.mapped_column(sa.String, unique=True, index=True)
    hashed_password: so.Mapped[str]
    is_active: so.Mapped[bool] = so.mapped_column(sa.Boolean, default=True)

    items: so.Mapped[list['Item']] = so.relationship(
        cascade='all, delete-orphan',
        back_populates='user'
    )

    def __repr__(self):
        return f'User({self.id})'

2. 修复会话操作逻辑 #

• 创建User后必须调用await db.refresh(db_user)，同步数据库生成的主键等字段，避免后续关联操作使用无效数据。
• 创建Item前验证User存在，避免传入无效user_id；同时确保owner_id正确传递。

修改main.py：

# main.py 修改后
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

import models, schemas
from database import SessionLocal

app = FastAPI()

async def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        await db.close()

@app.post("/users/", response_model=schemas.User)
async def create_user(user: schemas.UserCreate, db: AsyncSession = Depends(get_db)):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(
        username=user.username,
        email=user.email,
        hashed_password=fake_hashed_password
    )
    db.add(db_user)
    await db.commit()
    await db.refresh(db_user)  # 同步数据库生成的id等字段
    return db_user

@app.post("/users/{user_id}/items/", response_model=schemas.Item)
async def create_item_for_user(
    user_id: int,
    item: schemas.ItemCreate,
    db: AsyncSession = Depends(get_db)
):
    # 验证用户存在
    result = await db.execute(select(models.User).where(models.User.id == user_id))
    db_user = result.scalar_one_or_none()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    
    db_item = models.Item(
        title=item.title,
        description=item.description,
        owner_id=user_id
    )
    db.add(db_item)
    await db.commit()
    await db.refresh(db_item)
    return db_item

# 添加用户读取接口，验证修复效果
@app.get("/users/{user_id}", response_model=schemas.User)
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    # 显式加载关联的items数据
    result = await db.execute(
        select(models.User)
        .options(selectinload(models.User.items))
        .where(models.User.id == user_id)
    )
    db_user = result.scalar_one_or_none()
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

3. 异步模式关联数据加载说明 #

读取关联数据时，必须通过selectinload或joinedload显式指定加载策略，异步SQLAlchemy不会自动加载关联对象。

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