FastAPI+SQLAlchemy：如何在UserRequest中关联UserProfile并更新数据？

我正在使用FastAPI和SQLAlchemy，拥有两个带外键关联的ORM模型users和user_profiles，目前通过单个模型向这两个表插入数据。

ORM模型代码（简化版） #

# 第一个模型
class User():
    __tablename__: str = "user"
# 第二个模型
class UserProfile:
    __tablename__: str = "user_profiles"

更新用的请求模型 #

class UserRequest():
    user_id: str = someval()

我的问题是：如何在UserRequest中关联UserProfile模型并实现对应数据的更新？

1. 完善ORM模型的关联关系 #

先补全两个模型的字段与外键关联逻辑，确保User和UserProfile能正确映射：

from sqlalchemy import Column, String, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "user"
    user_id = Column(String, primary_key=True, index=True)
    # 建立与UserProfile的一对一关联
    profile = relationship("UserProfile", back_populates="user", uselist=False)

class UserProfile(Base):
    __tablename__ = "user_profiles"
    id = Column(String, primary_key=True, index=True)
    user_id = Column(String, ForeignKey("user.user_id"), unique=True)
    # 关联回User模型
    user = relationship("User", back_populates="profile")
    # 补充业务所需字段，比如昵称、手机号
    nickname = Column(String)
    phone = Column(String)

2. 修改UserRequest请求模型 #

用Pydantic模型嵌套UserProfile的更新字段，实现关联：

from pydantic import BaseModel

# 定义UserProfile的更新字段模型
class UserProfileUpdate(BaseModel):
    nickname: str | None = None
    phone: str | None = None

# 将Profile更新模型关联到UserRequest
class UserRequest(BaseModel):
    user_id: str
    profile: UserProfileUpdate | None = None

3. 实现FastAPI更新接口逻辑 #

通过SQLAlchemy会话找到用户，同步更新关联的UserProfile数据：

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
# 替换为你的模块导入路径
from your_db_module import get_db, User, UserProfile

app = FastAPI()

@app.put("/update-user/")
def update_user(user_request: UserRequest, db: Session = Depends(get_db)):
    # 查找目标用户
    db_user = db.query(User).filter(User.user_id == user_request.user_id).first()
    if not db_user:
        raise HTTPException(status_code=404, detail="用户不存在")
    
    # 处理Profile更新请求
    if user_request.profile:
        # 查找关联的Profile，不存在则根据业务选择创建或报错
        db_profile = db.query(UserProfile).filter(UserProfile.user_id == user_request.user_id).first()
        if not db_profile:
            # 自动创建关联的Profile（若业务允许）
            db_profile = UserProfile(user_id=user_request.user_id)
            db.add(db_profile)
        
        # 仅更新请求中提供的字段，避免覆盖原有数据
        update_data = user_request.profile.dict(exclude_unset=True)
        for key, value in update_data.items():
            setattr(db_profile, key, value)
    
    # 提交变更并刷新数据
    db.commit()
    db.refresh(db_user)
    return {"message": "更新成功", "user_info": db_user}

核心注意点 #

• exclude_unset=True确保只更新请求中明确传递的字段，保留未修改的原有数据
• 若业务要求必须先存在UserProfile才能更新，可移除自动创建逻辑，直接抛出404异常
• 一对一关系通过uselist=False保证一个用户仅对应一个Profile

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