SQLModel 数据库 ORM¶

为什么要学 SQLModel¶

SQLModel 是由 FastAPI 作者创建的 Python ORM 库，它将 Pydantic 和 SQLAlchemy 的优势合二为一。一个模型类同时充当数据库表定义、Pydantic 验证模型和 API 序列化模型。这消除了传统项目中需要维护 ORM 模型、Schema 模型和 API 模型三套代码的问题。对于 FastAPI 项目来说，SQLModel 是最自然的数据库方案。

核心概念¶

概念	白话解释	用途
SQLModel	模型基类	同时是 Pydantic Model 和 SQLAlchemy Model
Table	表标记	`table=True` 表示映射到数据库表
Field	字段	定义列属性和验证规则
Session	会话	数据库操作的上下文
Relationship	关系	表之间的关联
Engine	引擎	数据库连接

安装配置¶

pip install sqlmodel

# 常用数据库驱动
pip install psycopg2-binary  # PostgreSQL
pip install aiosqlite        # SQLite 异步
pip install asyncpg           # PostgreSQL 异步

快速上手¶

定义模型¶

from sqlmodel import SQLModel, Field, create_engine, Session, select
from typing import Optional

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    email: str = Field(unique=True)
    age: Optional[int] = Field(default=None, ge=0, le=150)
    is_active: bool = Field(default=True)

# 创建引擎和表
engine = create_engine("sqlite:///database.db", echo=True)
SQLModel.metadata.create_all(engine)

CRUD 操作¶

# 创建
with Session(engine) as session:
    user = User(name="Alice", email="alice@example.com", age=30)
    session.add(user)
    session.commit()
    session.refresh(user)
    print(user.id)  # 自动生成的 ID

# 查询
with Session(engine) as session:
    # 单条查询
    user = session.get(User, 1)

    # 条件查询
    statement = select(User).where(User.age >= 25)
    users = session.exec(statement).all()

    # 搜索
    statement = select(User).where(User.name.contains("Ali"))
    users = session.exec(statement).all()

# 更新
with Session(engine) as session:
    user = session.get(User, 1)
    user.age = 31
    session.add(user)
    session.commit()

# 删除
with Session(engine) as session:
    user = session.get(User, 1)
    session.delete(user)
    session.commit()

关系¶

from sqlmodel import Relationship

class Team(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    members: list["User"] = Relationship(back_populates="team")

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    team_id: Optional[int] = Field(default=None, foreign_key="team.id")
    team: Optional[Team] = Relationship(back_populates="members")

进阶用法¶

读写分离模型¶

# 基础模型（用于共享字段定义）
class UserBase(SQLModel):
    name: str
    email: str
    age: Optional[int] = None

# 数据库模型
class User(UserBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    hashed_password: str

# 创建请求模型
class UserCreate(UserBase):
    password: str

# 响应模型（不包含密码）
class UserRead(UserBase):
    id: int

# 更新模型
class UserUpdate(SQLModel):
    name: Optional[str] = None
    email: Optional[str] = None
    age: Optional[int] = None

FastAPI 集成¶

from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import Session, select

app = FastAPI()

def get_session():
    with Session(engine) as session:
        yield session

@app.post("/users/", response_model=UserRead)
def create_user(user: UserCreate, session: Session = Depends(get_session)):
    db_user = User.model_validate(user, update={"hashed_password": hash_pw(user.password)})
    session.add(db_user)
    session.commit()
    session.refresh(db_user)
    return db_user

@app.get("/users/", response_model=list[UserRead])
def read_users(skip: int = 0, limit: int = 100, session: Session = Depends(get_session)):
    users = session.exec(select(User).offset(skip).limit(limit)).all()
    return users

@app.get("/users/{user_id}", response_model=UserRead)
def read_user(user_id: int, session: Session = Depends(get_session)):
    user = session.get(User, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

异步支持¶

from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine

async_engine = create_async_engine("sqlite+aiosqlite:///database.db")

async def get_async_session():
    async with AsyncSession(async_engine) as session:
        yield session

@app.get("/users/")
async def read_users(session: AsyncSession = Depends(get_async_session)):
    result = await session.exec(select(User))
    return result.all()

复杂查询¶

from sqlmodel import select, col, or_, and_

# 排序和分页
statement = (
    select(User)
    .where(User.is_active == True)
    .order_by(User.name)
    .offset(0)
    .limit(10)
)

# OR 条件
statement = select(User).where(
    or_(User.age > 30, User.name == "Admin")
)

# JOIN
statement = (
    select(User, Team)
    .join(Team)
    .where(Team.name == "Engineering")
)
results = session.exec(statement).all()

常见问题¶

Q: 与 SQLAlchemy 的关系？¶

SQLModel 基于 SQLAlchemy 2.0，完全兼容。可以在同一项目中混用。

Q: 如何处理数据库迁移？¶

使用 Alembic：

pip install alembic
alembic init migrations
# 配置 env.py 使用 SQLModel.metadata
alembic revision --autogenerate -m "initial"
alembic upgrade head

Q: 性能如何？¶

与 SQLAlchemy 相同。Pydantic v2 带来的性能提升也适用于 SQLModel。

参考资源¶

官网：https://sqlmodel.tiangolo.com/
GitHub：https://github.com/tiangolo/sqlmodel
教程：https://sqlmodel.tiangolo.com/tutorial/
FastAPI 集成：https://sqlmodel.tiangolo.com/tutorial/fastapi/