FastAPI中返回SQLAlchemy ORM类的类型注解适配问题
解决方案
1. 正确定义SQLAlchemy ORM模型(满足类型检查)
使用SQLAlchemy 2.0+的Mapped和mapped_column定义模型,让类型检查工具(如mypy)能识别明确的类型:
from sqlalchemy import String, Integer from sqlalchemy.orm import DeclarativeBase, mapped_column, Mapped class Base(DeclarativeBase): pass class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(Integer, primary_key=True) name: Mapped[str] = mapped_column(String(50)) email: Mapped[str] = mapped_column(String(100), unique=True)
2. 定义适配ORM的Pydantic模型
创建对应的Pydantic模型,开启from_attributes=True(替代旧版orm_mode=True),支持直接从SQLAlchemy ORM实例读取属性:
from pydantic import BaseModel, ConfigDict class UserResponse(BaseModel): model_config = ConfigDict(from_attributes=True) id: int name: str email: str
3. 结合FastAPI的response_model与ORM类型注解
在FastAPI路由函数中,直接用ORM类作为返回类型注解,同时指定response_model为对应的Pydantic模型。既满足类型检查要求,又能让FastAPI正确序列化返回结果:
from fastapi import FastAPI, Depends from sqlalchemy.orm import Session from your_module import User, UserResponse, get_db # get_db为数据库会话依赖 app = FastAPI() @app.get("/users/{user_id}", response_model=UserResponse) def get_user(user_id: int, db: Session = Depends(get_db)) -> User: user = db.query(User).filter(User.id == user_id).first() return user
原理说明
- 类型检查工具会验证函数返回的确实是
User实例,符合团队代码审查的类型规范。 - FastAPI会忽略函数的返回类型注解,转而使用
response_model指定的Pydantic模型序列化返回值,自动将ORM实例转换为合法JSON格式,避免运行时报错。
进阶:通用装饰器简化重复代码
如果觉得每个路由都写response_model繁琐,可以封装通用装饰器,自动绑定ORM与对应Pydantic模型,同时保留ORM类型注解:
from functools import wraps from pydantic import TypeAdapter from fastapi import HTTPException def orm_to_pydantic(pydantic_model): def decorator(func): @wraps(func) def wrapper(*args, **kwargs): result = func(*args, **kwargs) if result is None: raise HTTPException(status_code=404, detail="Item not found") # 兼容单实例和列表返回 if isinstance(result, list): return TypeAdapter(list[pydantic_model]).validate_python(result) return TypeAdapter(pydantic_model).validate_python(result) # 保留原函数返回类型注解,确保类型检查通过 wrapper.__annotations__["return"] = func.__annotations__["return"] return wrapper return decorator
使用方式:
@app.get("/users/{user_id}") @orm_to_pydantic(UserResponse) def get_user(user_id: int, db: Session = Depends(get_db)) -> User: user = db.query(User).filter(User.id == user_id).first() return user
内容的提问来源于stack exchange,提问作者szw0407
相关产品推荐
相关产品推荐

