FastAPI-SQLModel关联配置：新增Email时自动创建VStatus行

现有表结构 #

当前使用SQLModel定义了两个关联数据表：

from typing import Optional, List
from sqlmodel import SQLModel, Field, Relationship
from enum import Enum

class StatusEnum(Enum):
    Unverified = "Unverified"
    InProgress = "InProgress"
    Verified = "Verified"

class Email(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    email: str
    vstatus_id: Optional[int] = Field(default=None, foreign_key='vstatus.id')
    vstatus: Optional["VStatus"] = Relationship()

class VStatus(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    vstatus: StatusEnum = Field(default=StatusEnum.Unverified)
    vdesc: str = Field(default="")

需求说明 #

当调用API接口/create_email，仅传入email字段创建新的Email数据时，需要通过SQLModel的Relationship自动在VStatus表中生成一条默认状态的新记录：

• 请求示例：

POST /create_email
Content-Type: application/json

{
    "email": "example@example.com"
}

• 预期响应：

{
    "email": "example@example.com",
    "vstatus_id": 1001
}

• 同时VStatus表自动新增行：

解决方案 #

1. 调整表结构的Relationship配置 #

修改两个模型的Relationship定义，添加级联创建规则并补充反向关联，确保创建Email时能自动触发VStatus的创建：

class Email(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    email: str
    vstatus_id: Optional[int] = Field(default=None, foreign_key='vstatus.id')
    # 配置级联规则，自动创建/维护关联的VStatus实例
    vstatus: Optional["VStatus"] = Relationship(
        sa_relationship_kwargs={"cascade": "all, delete-orphan"},
        back_populates="emails"
    )

class VStatus(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    vstatus: StatusEnum = Field(default=StatusEnum.Unverified)
    vdesc: str = Field(default="")
    # 补充反向关联，关联到Email表
    emails: List["Email"] = Relationship(back_populates="vstatus")

2. 自动初始化关联的VStatus实例 #

有两种方式实现自动创建VStatus：

方式一：给Email的vstatus字段设置默认工厂

直接在模型中配置，创建Email时自动生成默认VStatus实例：

class Email(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    email: str
    vstatus_id: Optional[int] = Field(default=None, foreign_key='vstatus.id')
    vstatus: Optional["VStatus"] = Relationship(
        sa_relationship_kwargs={"cascade": "all, delete-orphan"},
        back_populates="emails",
        default_factory=VStatus  # 默认创建VStatus实例
    )

方式二：在API逻辑中主动初始化

如果不想修改模型默认值，可在业务代码中手动关联新的VStatus实例：

from sqlmodel import Session, create_engine

# 数据库连接示例
engine = create_engine("sqlite:///database.db")
SQLModel.metadata.create_all(engine)

def create_email(email_data: dict):
    with Session(engine) as session:
        # 创建Email时主动关联新的VStatus实例
        new_email = Email(
            email=email_data["email"],
            vstatus=VStatus()
        )
        session.add(new_email)
        session.commit()
        session.refresh(new_email)
        return {
            "email": new_email.email,
            "vstatus_id": new_email.vstatus_id
        }

3. API接口实现（以FastAPI为例） #

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class EmailCreateRequest(BaseModel):
    email: str

@app.post("/create_email")
def handle_create_email(request: EmailCreateRequest):
    result = create_email(request.dict())
    return result

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