Denny的博客

Denny的博客

0%

FastAPI自学教程（42） - SQL数据库集成

发表于 更新于 分类于

1.基础用法

1
2
3
4
5
6
7
8
9
10
11
12
13
from sqlmodel import SQLModel, Field, create_engine, Session

class Hero(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    age: int | None = None
    secret_name: str

sqlite_url = "sqlite:///database.db"
engine = create_engine(sqlite_url)

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)
  • 使用SQLModel结合Pydantic和SQLAlchemy实现ORM映射
  • 通过table=True声明数据库表模型，字段类型自动映射为数据库列类型
  • 创建数据库引擎时支持SQLite/PostgreSQL等数据库协议

流程图： 

graph TD
    A[客户端请求] --> B[创建数据库会话]
    B --> C[执行CRUD操作]
    C --> D[提交事务]
    D --> E[关闭会话]
    E --> F[返回响应]

2.数据库会话管理

1
2
3
4
5
6
7
8
9
10
11
12
from fastapi import Depends
from sqlmodel import Session

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

SessionDep = Annotated[Session, Depends(get_session)]

@app.on_event("startup")
def on_startup():
    create_db_and_tables()
  • 使用yield生成器创建会话依赖项，确保每个请求独立数据库会话
  • 通过@app.on_event("startup")在应用启动时自动建表
  • 会话生命周期管理包含自动提交和回滚机制

3.CRUD操作实现

3.1 创建数据

1
2
3
4
5
6
@app.post("/heroes/")
def create_hero(hero: Hero, session: SessionDep):
    session.add(hero)
    session.commit()
    session.refresh(hero)
    return hero
  • session.add()将对象加入会话追踪队列
  • session.commit()持久化数据到数据库
  • session.refresh()获取自增ID等数据库生成值

3.2 查询数据

1
2
3
4
5
6
@app.get("/heroes/{hero_id}")
def read_hero(hero_id: int, session: SessionDep):
    hero = session.get(Hero, hero_id)
    if not hero:
        raise HTTPException(404, "Hero not found")
    return hero
  • 支持分页查询select().offset().limit()
  • 通过Field(index=True)优化查询性能

4.高级特性

4.1 模型继承

1
2
3
4
5
6
class Person(SQLModel):
    name: str = Field(index=True)
    age: int

class Hero(Person, table=True):
    secret_name: str
  • 支持模型继承实现字段复用
  • 基类可同时用于数据验证和数据库表结构

4.2 生产环境配置

1
2
3
# PostgreSQL配置示例
postgres_url = "postgresql://user:password@localhost:5432/mydb"
engine = create_engine(postgres_url, pool_size=20, max_overflow=0)
  • 推荐使用连接池管理数据库连接
  • 通过Alembic实现数据库迁移管理

5.注意事项

  1. 线程安全：SQLite需配置connect_args={"check_same_thread": False}
  2. 事务管理：复杂操作应使用with session.begin()事务块
  3. 性能优化：避免N+1查询问题，使用selectinload预加载关联数据
  4. 环境隔离：开发/测试/生产环境使用不同数据库配置

测试命令示例： 

1
2
# 创建新英雄
curl -X POST -H "Content-Type: application/json" -d '{"name":"Deadpond","secret_name":"Dive Wilson"}' http://localhost:8000/heroes/