FastAPI+SQLAlchemy多对多关联表POST添加记录问题求助

问题背景 #

已有Pydantic与SQLAlchemy的多对多模型（游戏与用户表通过中间表关联），需编写POST接口实现两者关联记录的添加，但现有接口代码无法正常运行，需修正实现。

模型代码 #

from sqlalchemy import Table, Column, Integer, String, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from pydantic import BaseModel, Query
from typing import List

Base = declarative_base()

game_users = Table('game_users', Base.metadata,
                   Column('game_id', ForeignKey('games.id'), primary_key=True),
                   Column('user_id', ForeignKey('users.id'), primary_key=True)
                   )


class Game(Base):
    __tablename__ = 'games'
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    users = relationship("User", secondary="game_users", back_populates='games')


class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    age = Column(Integer)
    email = Column(String, nullable=False, unique=True)
    games = relationship("Game", secondary="game_users", back_populates='users')


class UserBase(BaseModel):
    id: int
    name: str
    age: int = Query(ge=0, le=100)
    email: str

    class Config:
        orm_mode = True


class GameBase(BaseModel):
    id: int
    name: str

    class Config:
        orm_mode = True


class UsersOut(UserBase):
    games: List[GameBase]


class GamesOut(GameBase):
    users: List[UserBase]

原错误接口代码 #

@app.post('/connect/{uid}/{gid}')
def connect_to_game(uid: int, gid: int, db: Session = Depends(get_db)):
    game = db.query(Game).filter(Game.id == gid).first()
    user = db.query(User).filter(User.id == uid).first()

    user_games = user(games=[game.id])
    game_users = game(users=[user.id])

    db.add_all([user_games, game_users])
    db.commit()

    return f'{game.name} successfully connected to {user.name}'

问题分析 #

1. ORM实例调用错误：user(games=[game.id])是错误写法，user是数据库查询返回的ORM实例，不是类，不能通过调用方式修改属性。
2. 多对多关联操作逻辑错误：SQLAlchemy的多对多关联可直接通过实例的relationship属性（user.games/game.users）操作，无需创建新实例或调用db.add，因为查询出的实例已处于数据库会话中。
3. 缺失异常处理：未处理用户或游戏不存在的情况，会导致后续代码抛出未捕获的异常。

修正后的接口代码 #

from fastapi import HTTPException

@app.post('/connect/{uid}/{gid}')
def connect_to_game(uid: int, gid: int, db: Session = Depends(get_db)):
    # 查询目标用户与游戏
    game = db.query(Game).filter(Game.id == gid).first()
    user = db.query(User).filter(User.id == uid).first()

    # 处理资源不存在的情况
    if not user:
        raise HTTPException(status_code=404, detail=f"用户ID {uid} 不存在")
    if not game:
        raise HTTPException(status_code=404, detail=f"游戏ID {gid} 不存在")
    
    # 检查是否已关联，避免重复添加触发主键冲突
    if game not in user.games:
        user.games.append(game)
        db.commit()
        db.refresh(user)  # 可选：刷新实例以获取最新关联数据
    
    return f'{game.name} 已成功关联到 {user.name}'

关键说明 #

• 直接操作relationship属性：通过user.games.append(game)即可向中间表添加关联记录，SQLAlchemy会自动处理中间表的写入逻辑。
• 重复关联检查：添加前验证关联是否已存在，避免中间表出现重复主键记录导致数据库异常。
• 异常处理：通过HTTPException返回明确的404错误，提升接口的健壮性和可读性。

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