连接数据库
前言
FastAPI 本身不绑定数据库，一般通过 ORM（如 SQLAlchemy）操作关系型数据库，通过 redis-py 操作 Redis。本章整理：
- SQLAlchemy 同步用法
- SQLAlchemy 异步用法（async engine + async session，推荐）
- 通过依赖注入管理会话生命周期
- Redis 常用操作
连接关系型数据库（SQLAlchemy）
模型与请求体要分开
危险
常见错误是把 Pydantic 模型直接当作 ORM 模型使用（如 session.add(pydantic对象)），这是不对的：Pydantic 负责请求/响应校验，SQLAlchemy 模型负责数据库映射，两者需要分开定义。
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.orm import declarative_base, sessionmaker
Base = declarative_base()
class Item(Base):
"""SQLAlchemy ORM 模型"""
__tablename__ = "items"
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(64), nullable=False)
description = Column(String(255), default="")
engine = create_engine("mysql+mysqlconnector://user:password@localhost/testdb")
SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False)
Base.metadata.create_all(bind=engine) # 演示用，生产环境建议用 Alembic 迁移
同步 + 依赖注入
注意
不要在模块级别创建全局 session 供所有请求共用（会有并发与事务问题），正确的做法是每个请求一个会话，用依赖注入管理生命周期。
from typing import Generator
from fastapi import Depends, FastAPI
from pydantic import BaseModel
from sqlalchemy.orm import Session
app = FastAPI()
# Pydantic 请求体模型（与 ORM 模型分开）
class ItemCreate(BaseModel):
name: str
description: str | None = None
# 依赖注入：每个请求获取一个 session，请求结束自动关闭
def get_db() -> Generator[Session, None, None]:
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.post("/items/")
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
db_item = Item(name=item.name, description=item.description)
db.add(db_item)
db.commit()
db.refresh(db_item)
return {"id": db_item.id, "name": db_item.name}
@app.get("/items/{item_id}")
def read_item(item_id: int, db: Session = Depends(get_db)):
item = db.get(Item, item_id)
if item is None:
from fastapi import HTTPException
raise HTTPException(status_code=404, detail="Item not found")
return {"id": item.id, "name": item.name, "description": item.description}
异步 SQLAlchemy
FastAPI 推荐异步写法。使用 create_async_engine + async_sessionmaker + AsyncSession：
python -m pip install sqlalchemy aiosqlite
# MySQL 异步驱动: pip install asyncmy（或 aiomysql）
from contextlib import asynccontextmanager
from typing import AsyncGenerator
from fastapi import Depends, FastAPI
from pydantic import BaseModel
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.asyncio import (
AsyncSession,
async_sessionmaker,
create_async_engine,
)
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(64), nullable=False)
# SQLite 异步示例；MySQL 替换为 create_async_engine("mysql+asyncmy://user:password@host/db")
engine = create_async_engine("sqlite+aiosqlite:///./test.db", echo=False)
AsyncSessionLocal = async_sessionmaker(
engine,
class_=AsyncSession,
expire_on_commit=False,
autoflush=False,
)
async def init_tables():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
@asynccontextmanager
async def lifespan(app: FastAPI):
await init_tables()
yield
await engine.dispose()
@asynccontextmanager
async def get_db_session() -> AsyncGenerator[AsyncSession, None]:
"""异步会话依赖，自动提交/回滚/关闭"""
async with AsyncSessionLocal() as session:
try:
yield session
except Exception:
await session.rollback()
raise
app = FastAPI(lifespan=lifespan)
class ItemCreate(BaseModel):
name: str
@app.post("/items/")
async def create_item(item: ItemCreate, db: AsyncSession = Depends(get_db_session)):
db_item = Item(name=item.name)
db.add(db_item)
await db.commit()
await db.refresh(db_item)
return {"id": db_item.id, "name": db_item.name}
提示
异步会话要点：expire_on_commit=False 避免提交后对象属性失效；查询用 await db.execute(select(...))；提交后如需访问对象属性，用 await db.refresh(obj)。
连接 Redis
本示例环境：Python 3.11+、Redis 7.x。安装：
python -m pip install redis
连接池与启动/关闭
把 redis 客户端放在单独模块，并在应用启动时建立连接、停止时关闭：
├── demo.py
└── pkg
├── __init__.py
└── redis
└── __init__.py
pkg/redis/__init__.py：
from redis import ConnectionPool, Redis
# 连接池，max_connections 限制最大连接数
pool = ConnectionPool(
host="192.168.1.112",
port=6379,
db=1,
password="123456",
max_connections=10,
decode_responses=True, # 返回 str 而不是 bytes
)
redis_client = Redis(connection_pool=pool)
__all__ = ["redis_client"]
demo.py：
from contextlib import asynccontextmanager
from datetime import datetime
import uvicorn
from fastapi import FastAPI
from pkg.redis import redis_client
@asynccontextmanager
async def applifespan(app: FastAPI):
uptime = datetime.now()
redis_client.set("uptime", uptime.strftime("%Y-%m-%d %H:%M:%S"))
try:
yield
finally:
redis_client.close()
app = FastAPI(lifespan=applifespan)
@app.get("/")
async def get_root():
uptime = redis_client.get("uptime")
return {"msg": f"uptime is {uptime}"}
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
常见操作
# 检测连接是否正常
redis_client.ping()
# 字符串 set / get / delete / exists
redis_client.set("k1", "v1")
redis_client.get("k1")
redis_client.delete("k1")
redis_client.exists("k1") # 是否存在
# 过期时间
redis_client.setex("greeting", 10, "hello world") # ttl 为 10 秒
redis_client.ttl("greeting") # 查看剩余 ttl
redis_client.expire("greeting", 100) # 刷新 ttl 为 100 秒
# 批量设置字典
dict_data = {
"employee_name": "Adam Adams",
"employee_age": 30,
"position": "Software Engineer",
}
redis_client.mset(dict_data)
# 自增：incr 默认 +1，incrby 指定步长
redis_client.set("request_total", 0)
redis_client.incr("request_total") # +1
redis_client.incrby("request_total", 10) # +10
# 管道：批量操作，减少多次网络往返
with redis_client.pipeline() as pipe:
for _ in range(10000):
pipe.incr("request_total")
pipe.execute() # 统一提交
# 列表：lpush -> rpop 是队列（FIFO），lpush -> lpop 是栈（FILO）
redis_client.lpush("language", "python")
redis_client.lpush("language", "go")
redis_client.lpush("language", "rust")
redis_client.rpop("language") # python
redis_client.rpop("language") # go
redis_client.rpop("language") # rust
redis_client.llen("language") # 列表长度
# 集合
redis_client.sadd("warframe", "atlas")
redis_client.sadd("warframe", "banshee")
redis_client.srem("warframe", "atlas") # 删除元素
注意
易错点：get(key) 只有读取一个参数，指定步长自增要用 incrby(key, amount)；管道提交是 pipe.execute()。