FastAPI + SQLModel 实战：标准项目结构下，一个模型搞定数据库与 API

零重复代码！用 SQLModel 实现 数据库模型 与 Pydantic 请求/响应模型 的统一，同时保留字段校验、类型提示与文档自动生成。

🎯 为什么 SQLModel 能“一个模型走天下”？

在传统 FastAPI 项目中，你通常要写：

• UserCreate（请求体）
• UserRead（响应体）
• UserUpdate（部分更新）
• UserDB（SQLAlchemy ORM 模型）

而 SQLModel 的核心优势是：

✅ 继承自 Pydantic 的 BaseModel → 支持字段校验、类型提示、OpenAPI 文档
✅ 同时是 SQLAlchemy 的 ORM 模型 → 可直接用于数据库操作
✅ 一套代码，双重身份 → 无需重复定义！

📁 标准 FastAPI 项目结构

fastapi-sqlmodel-user/
├── main.py
├── database.py
├── models/
│   └── user.py
├── api/
│   └── v1/
│       └── users.py
└── requirements.txt

1️⃣ 定义 唯一 的 User 模型（models/user.py）

# models/user.py
from typing import Optional
from sqlmodel import SQLModel, Field

class User(SQLModel, table=True):
    """
    这个类既是：
    - 数据库表（通过 table=True）
    - Pydantic 模型（用于 API 输入/输出）
    - 自动支持字段校验（如必填、类型、约束）
    """
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(min_length=1, max_length=100)  # Pydantic 校验生效！
    email: str = Field(unique=True, regex=r"^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$")
    age: Optional[int] = Field(default=None, ge=0, le=150)  # 年龄范围校验

🔥 关键点：

• Field() 来自 Pydantic，所以 min_length, regex, ge 等校验全部生效
• unique=True 是 SQLAlchemy 的数据库约束
• 同一个 Field 同时服务 API 层 和 DB 层

2️⃣ 数据库配置（database.py）

# database.py
from sqlmodel import create_engine, Session
from sqlmodel import SQLModel
from models.user import User  # 确保模型被导入

# 使用 SQLite 文件数据库（持久化）
engine = create_engine("sqlite:///./app.db", echo=True)

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

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

3️⃣ API 路由（api/v1/users.py）

# api/v1/users.py
from fastapi import APIRouter, Depends, HTTPException
from sqlmodel import Session, select
from models.user import User
from database import get_session

router = APIRouter(prefix="/v1/users", tags=["users"])

@router.post("/", response_model=User)
def create_user(user: User, session: Session = Depends(get_session)):
    """
    请求体自动用 User 模型校验！
    - name 必填且 1~100 字符
    - email 必须是合法邮箱
    - age 必须在 0~150 之间
    """
    # 检查邮箱是否已存在（unique 约束在 DB 层，但提前校验更友好）
    existing_user = session.exec(select(User).where(User.email == user.email)).first()
    if existing_user:
        raise HTTPException(status_code=400, detail="Email already registered")

    session.add(user)
    session.commit()
    session.refresh(user)  # 获取数据库生成的 id
    return user


@router.get("/", response_model=list[User])
def read_users(session: Session = Depends(get_session)):
    return session.exec(select(User)).all()


@router.get("/{user_id}", response_model=User)
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

4️⃣ 主应用入口（main.py）

# main.py
from fastapi import FastAPI
from database import create_db_and_tables, engine
from api.v1.users import router as users_router

app = FastAPI(title="FastAPI + SQLModel 用户管理")

# 初始化数据库
@app.on_event("startup")
def on_startup():
    create_db_and_tables()

# 注册路由
app.include_router(users_router)

5️⃣ 运行 & 测试

pip install fastapi sqlmodel uvicorn[standard]
uvicorn main:app --reload

访问 http://localhost:8000/docs

✅ 测试字段校验

1. 邮箱格式错误（自动拒绝）

POST /v1/users/
{
   
  "name": "张三",
  "email": "invalid-email",
  "age": 30
}

→ 返回 422 错误："msg": "string does not match regex ..."

2. 年龄超范围

{
   
  "name": "李四",
  "email": "li@example.com",
  "age": 200
}

→ 返回 422："msg": "ensure this value is less than or equal to 150"

3. 成功创建

{
   
  "name": "王五",
  "email": "wang@example.com",
  "age": 25
}

→ 返回带 id 的完整用户对象

💡 为什么这比传统方式好？