如何在FastAPI路由中使用SQLAlchemy的Base派生类？

我定义了SQLAlchemy的User类，希望在FastAPI路由中直接使用它，避免重复创建结构完全相同的BaseModel派生类：

class User(Base):
    __tablename__ = "users"

    id = Column(UUID(as_uuid=True), default=uuid.uuid4, primary_key=True)

    username = Column(String(100), nullable=False, unique=True)

    created_on = Column(DateTime(), default=datetime.now)
    updated_on = Column(DateTime(), default=datetime.now, onupdate=datetime.now)

尝试在路由中直接使用该类：

@app.post("/user")
async def create_user(user: User) -> Any:
    return None

但出现如下错误：

fastapi.exceptions.FastAPIError: Invalid args for response field! Hint: check that <class 'db.User'> is a valid Pydantic field type.

请问如何在FastAPI路由中使用SQLAlchemy的Base派生类？是否可行？是否必须重复创建BaseModel派生类才能用于FastAPI路由？

解答 #

直接将SQLAlchemy的Base派生类用作FastAPI的请求参数或响应模型不可行，因为FastAPI依赖Pydantic模型完成数据校验、序列化/反序列化等核心工作，而SQLAlchemy ORM类不具备这些特性，因此会触发报错。不过我们有多种方式避免重复定义结构，以下是具体方案：

1. 使用SQLModel（推荐） #

SQLModel是FastAPI作者开发的库，直接整合了SQLAlchemy ORM和Pydantic模型的功能，定义一次类即可同时作为数据库模型和API的请求/响应模型，彻底解决重复定义问题。

示例代码：

from sqlmodel import SQLModel, Field
from uuid import UUID, uuid4
from datetime import datetime
from fastapi import FastAPI

app = FastAPI()

class User(SQLModel, table=True):
    __tablename__ = "users"
    id: UUID = Field(default_factory=uuid4, primary_key=True)
    username: str = Field(max_length=100, nullable=False, unique=True)
    created_on: datetime = Field(default_factory=datetime.now)
    updated_on: datetime = Field(default_factory=datetime.now, sa_column_kwargs={"onupdate": datetime.now})

@app.post("/user")
async def create_user(user: User) -> User:
    # 直接将User作为SQLAlchemy模型进行数据库操作
    db_user = User.model_validate(user)
    # 此处添加数据库写入逻辑
    return db_user

2. 自动从SQLAlchemy类生成Pydantic模型 #

如果不想替换现有SQLAlchemy代码，可以使用pydantic-sqlalchemy库自动生成对应Pydantic模型，无需手动重复字段定义。

首先安装依赖：

pip install pydantic-sqlalchemy

示例代码：

from pydantic_sqlalchemy import sqlalchemy_to_pydantic
from sqlalchemy import Column, String, DateTime, UUID
from sqlalchemy.ext.declarative import declarative_base
from uuid import uuid4, UUID
from datetime import datetime
from fastapi import FastAPI

Base = declarative_base()
app = FastAPI()

class User(Base):
    __tablename__ = "users"
    id = Column(UUID(as_uuid=True), default=uuid4, primary_key=True)
    username = Column(String(100), nullable=False, unique=True)
    created_on = Column(DateTime(), default=datetime.now)
    updated_on = Column(DateTime(), default=datetime.now, onupdate=datetime.now)

# 自动生成Pydantic模型
UserPydantic = sqlalchemy_to_pydantic(User)

@app.post("/user")
async def create_user(user: UserPydantic) -> UserPydantic:
    # 将Pydantic模型转换为SQLAlchemy实例
    db_user = User(**user.model_dump())
    # 数据库操作逻辑
    return db_user

3. 手动抽离字段定义复用 #

如果不想引入第三方库，可以把字段定义抽离出来，同时用于SQLAlchemy类和Pydantic模型：

from sqlalchemy import Column, String, DateTime, UUID
from sqlalchemy.ext.declarative import declarative_base
from pydantic import BaseModel
from uuid import uuid4, UUID
from datetime import datetime
from fastapi import FastAPI

Base = declarative_base()
app = FastAPI()

# 抽离共用的字段配置
USER_FIELD_CONFIGS = {
    "id": {
        "type": UUID,
        "sqlalchemy_column": Column(UUID(as_uuid=True), default=uuid4, primary_key=True)
    },
    "username": {
        "type": str,
        "sqlalchemy_column": Column(String(100), nullable=False, unique=True)
    },
    "created_on": {
        "type": datetime,
        "sqlalchemy_column": Column(DateTime(), default=datetime.now)
    },
    "updated_on": {
        "type": datetime,
        "sqlalchemy_column": Column(DateTime(), default=datetime.now, onupdate=datetime.now)
    }
}

# SQLAlchemy模型
class User(Base):
    __tablename__ = "users"
    id = USER_FIELD_CONFIGS["id"]["sqlalchemy_column"]
    username = USER_FIELD_CONFIGS["username"]["sqlalchemy_column"]
    created_on = USER_FIELD_CONFIGS["created_on"]["sqlalchemy_column"]
    updated_on = USER_FIELD_CONFIGS["updated_on"]["sqlalchemy_column"]

# Pydantic模型
class UserModel(BaseModel):
    id: UUID
    username: str
    created_on: datetime
    updated_on: datetime

    class Config:
        orm_mode = True

@app.post("/user")
async def create_user(user: UserModel) -> UserModel:
    db_user = User(**user.model_dump())
    # 数据库操作逻辑
    return db_user

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