FastAPI ORM模式与数据模型转换教程

1. 引言

FastAPI 是一个现代、快速的 Python Web 框架，广泛用于构建 API。在实际开发中，经常需要与数据库交互，ORM（对象关系映射）是常用的工具。数据模型转换则指如何在 API 层（使用 Pydantic 模型）和数据库层（使用 ORM 模型）之间无缝传递数据。本教程将引导您从基础到进阶，掌握 FastAPI 中的 ORM 和数据模型转换，让您的 API 开发更高效、更易维护。

2. 什么是 ORM？

ORM（Object-Relational Mapping）是一种编程技术，允许您使用面向对象的方式操作数据库，而无需直接编写 SQL 查询。它将数据库表映射为 Python 类，行映射为对象，使代码更易读和维护。在 FastAPI 中，ORM 有助于简化数据库交互，常见的 ORM 库包括 SQLAlchemy 和 Tortoise-ORM。

3. FastAPI 中的 ORM 选择

FastAPI 本身不捆绑特定 ORM，开发者可以自由选择。推荐使用 SQLAlchemy，因为它成熟、功能强大且与 Pydantic 兼容性好。本教程将以 SQLAlchemy 为例进行讲解。

4. 定义 ORM 模型

首先，我们需要定义数据库表对应的 ORM 模型。以下是一个示例，假设我们有一个 users 表。

# 导入必要的库
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 创建基础类
Base = declarative_base()

# 定义 User ORM 模型
class User(Base):
    __tablename__ = "users"
    
    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, nullable=False)
    email = Column(String, unique=True, nullable=False)
    age = Column(Integer, nullable=True)

# 配置数据库连接
DATABASE_URL = "sqlite:///./test.db"  # 以 SQLite 为例，可替换为其他数据库
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False} if DATABASE_URL.startswith("sqlite") else {})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 创建表
Base.metadata.create_all(bind=engine)

这个示例定义了一个 User 类，映射到数据库中的 users 表，并配置了 SQLite 数据库连接。

5. 定义 Pydantic 数据模型

Pydantic 是 FastAPI 的核心组件，用于数据验证和序列化。我们需要定义 Pydantic 模型来表示 API 的请求和响应数据。

from pydantic import BaseModel, EmailStr
from typing import Optional

# 定义 Pydantic 模型
class UserCreate(BaseModel):  # 用于创建用户的请求
    username: str
    email: EmailStr  # 使用 EmailStr 进行邮箱格式验证
    age: Optional[int] = None  # 可选字段

class UserResponse(BaseModel):  # 用于响应的用户数据
    id: int
    username: str
    email: str
    age: Optional[int]

    class Config:
        orm_mode = True  # 启用 ORM 模式，允许从 ORM 对象转换

注意：UserResponse 模型中的 Config 类设置了 orm_mode = True，这是实现转换的关键。

6. ORM 模型与 Pydantic 模型的转换方法

在 FastAPI 中，转换通常通过 Pydantic 的 from_orm 方法或 dict() 转换实现。以下介绍两种常用方法。

方法 1：使用 from_orm 方法（推荐）

当 Pydantic 模型启用了 orm_mode 后，可以从 ORM 对象直接创建 Pydantic 实例。

# 示例：从 ORM 对象转换到 Pydantic 模型
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/users/", response_model=UserResponse)
async def create_user(user: UserCreate, db: Session = Depends(get_db)):
    # 将 Pydantic 模型转换为字典，然后创建 ORM 对象
    db_user = User(**user.dict())
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    
    # 使用 from_orm 将 ORM 对象转换为 Pydantic 响应模型
    return UserResponse.from_orm(db_user)

方法 2：使用自定义转换

如果需要更复杂的转换，可以在模型中使用自定义方法或函数。

# 示例：自定义转换函数
def convert_orm_to_pydantic(orm_user: User) -> dict:
    return {
        "id": orm_user.id,
        "username": orm_user.username,
        "email": orm_user.email,
        "age": orm_user.age
    }

@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user is None:
        return {"error": "User not found"}
    # 使用自定义转换
    return UserResponse(**convert_orm_to_pydantic(db_user))

7. 实战示例：CRUD 操作

结合上述知识，实现一个完整的 CRUD API。

@app.get("/users/", response_model=list[UserResponse])
async def get_users(db: Session = Depends(get_db)):
    users = db.query(User).all()
    return [UserResponse.from_orm(user) for user in users]

@app.put("/users/{user_id}", response_model=UserResponse)
async def update_user(user_id: int, user_update: UserCreate, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user is None:
        return {"error": "User not found"}
    for key, value in user_update.dict().items():
        setattr(db_user, key, value)
    db.commit()
    db.refresh(db_user)
    return UserResponse.from_orm(db_user)

@app.delete("/users/{user_id}")
async def delete_user(user_id: int, db: Session = Depends(get_db)):
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user is None:
        return {"error": "User not found"}
    db.delete(db_user)
    db.commit()
    return {"message": "User deleted"}

8. 最佳实践和常见问题

• 保持模型简洁：ORM 模型只定义数据库字段，Pydantic 模型处理验证和序列化。
• 使用 orm_mode：这是 Pydantic 提供的便捷功能，简化转换过程。
• 处理嵌套模型：如果涉及关系（如外键），可以在 Pydantic 模型中定义嵌套模型。
• 错误处理：在实际应用中，添加适当的错误处理，如数据库连接失败或数据验证失败。
• 性能考虑：在大型应用中，避免 N+1 查询问题，可以通过 ORM 的懒加载或预加载优化。

9. 总结

通过本教程，您学会了在 FastAPI 中如何使用 ORM（以 SQLAlchemy 为例）进行数据库操作，并通过 Pydantic 模型实现数据转换。关键点包括：定义 ORM 和 Pydantic 模型、使用 from_orm 方法进行转换，以及构建完整的 CRUD API。练习这些概念，您将能更高效地开发 FastAPI 应用。

如需进一步学习，建议探索 FastAPI 官方文档和 SQLAlchemy 教程。祝您学习顺利！