20. FastAPI 集成 SQLAlchemy 2.x Async 与 aiomysql

安装依赖

uv add "sqlalchemy[asyncio]>=2,<3" aiomysql "pydantic-settings>=2,<3"

一、配置管理：集中读取主从库信息

数据库地址、账号和密码放在 .env 中，不把真实密码提交到版本控制。

1. .env 文件示例

# 主库：写入以及要求强一致的读取
DB_HOST=127.0.0.1
DB_PORT=3306
DB_USER=app_user
DB_PASSWORD=app_pass_123
DB_NAME=app_db

# 可选从库：允许复制延迟的查询
DB_SLAVE_HOST=127.0.0.1
DB_SLAVE_PORT=3307
DB_SLAVE_USER=app_readonly
DB_SLAVE_PASSWORD=readonly_pass
DB_SLAVE_NAME=app_db

.env 必须加入 .gitignore，只提交不含真实密码的 .env.example。

2. config.py

from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    db_host: str = "127.0.0.1"
    db_port: int = 3306
    db_user: str
    db_password: str
    db_name: str

    db_slave_host: str | None = None
    db_slave_port: int | None = None
    db_slave_user: str | None = None
    db_slave_password: str | None = None
    db_slave_name: str | None = None

    model_config = SettingsConfigDict(env_file=".env", extra="ignore")


settings = Settings()

配置在应用启动时读取一次。数据库密码如果包含 @、:、/ 等字符，不应直接手动拼 URL，后面的 URL.create() 会负责正确转义。

二、创建异步引擎和 AsyncSession

1. aiomysql 连接 URL

异步 MySQL URL 的格式为：

mysql+aiomysql://用户:密码@主机:端口/数据库?charset=utf8mb4

其中 mysql 是 SQLAlchemy 方言，aiomysql 是异步 DBAPI 驱动。

2. database.py：主库引擎和会话工厂

from sqlalchemy import URL
from sqlalchemy.ext.asyncio import (
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from config import settings


def make_url(
    host: str,
    port: int,
    user: str,
    password: str,
    name: str,
) -> URL:
    return URL.create(
        "mysql+aiomysql",
        username=user,
        password=password,
        host=host,
        port=port,
        database=name,
        query={"charset": "utf8mb4"},
    )


MASTER_URL = make_url(
    settings.db_host,
    settings.db_port,
    settings.db_user,
    settings.db_password,
    settings.db_name,
)

master_engine: AsyncEngine = create_async_engine(
    MASTER_URL,
    echo=False,          # 调试时改为 True，可打印 SQL
    pool_pre_ping=True,  # 取连接时检查失效连接
    pool_recycle=3600,   # 定期回收连接，避免 MySQL 空闲超时
    pool_size=5,
    max_overflow=10,
)

# 主库会话：后续写入示例统一使用 SessionLocal
SessionLocal = async_sessionmaker(
    bind=master_engine,
    class_=AsyncSession,
    autoflush=False,
    expire_on_commit=False,
)

关键点：

• AsyncEngine 是引擎类型；真正创建引擎的是 create_async_engine()。
• 创建引擎不等于立即连接 MySQL，首次执行数据库操作时才会使用连接池。
• async_sessionmaker 创建的是工厂，不是全局共享 Session。
• autoflush=False 表示查询前不自动 flush；expire_on_commit=False 方便提交后读取已经加载的属性。
• pool_pre_ping 和 pool_recycle 在异步引擎中仍然有效，连接检查由 SQLAlchemy 在异步流程中完成。

3. 声明式模型不需要异步化

# models.py
from sqlalchemy import Boolean, String, true
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"
    __table_args__ = {"mysql_engine": "InnoDB", "mysql_charset": "utf8mb4"}

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(50), nullable=False)
    email: Mapped[str] = mapped_column(String(254), unique=True)
    active: Mapped[bool] = mapped_column(
        Boolean,
        default=True,
        server_default=true(),
    )
    batch_tag: Mapped[str | None] = mapped_column(String(32), index=True)

这些字段与第 21 篇完整关系模型保持一致，第 21 篇会继续补充 Profile、Post、Role 和关联关系。

模型只描述表和对象结构。真正需要 await 的是引擎连接和 Session 数据库操作。

三、可选读写分离：显式选择异步会话

本篇采用两个会话工厂，不再通过同步 Session.get_bind() 做隐式路由：

• SessionLocal 绑定主库，负责写入和强一致读取。
• ReadSessionLocal 优先绑定从库，负责允许复制延迟的读取。
• 没有配置从库时，ReadSessionLocal 自动回退到主库。

在 database.py 后追加：

slave_engine: AsyncEngine | None = None

if settings.db_slave_host:
    slave_options = (
        settings.db_slave_port,
        settings.db_slave_user,
        settings.db_slave_password,
        settings.db_slave_name,
    )
    if any(value is None for value in slave_options):
        raise ValueError("启用从库时，端口、账号、密码和库名都必须配置")

    SLAVE_URL = make_url(
        settings.db_slave_host,
        settings.db_slave_port,
        settings.db_slave_user,
        settings.db_slave_password,
        settings.db_slave_name,
    )
    slave_engine = create_async_engine(
        SLAVE_URL,
        echo=False,
        pool_pre_ping=True,
        pool_recycle=3600,
        pool_size=5,
        max_overflow=10,
    )

ReadSessionLocal = async_sessionmaker(
    bind=slave_engine or master_engine,
    class_=AsyncSession,
    autoflush=False,
    expire_on_commit=False,
)

这里不是“同一会话自动识别 SELECT 和 INSERT”。调用方必须明确选择会话，避免事务中途切换数据库。刚写完立刻读取、需要读取最新状态的操作仍应走主库；从库可能存在复制延迟。

四、FastAPI 异步依赖注入与生命周期

1. deps.py

from collections.abc import AsyncIterator

from sqlalchemy.ext.asyncio import AsyncSession

from database import ReadSessionLocal, SessionLocal


async def get_db() -> AsyncIterator[AsyncSession]:
    """读取会话：优先从库，请求结束自动关闭。"""
    async with ReadSessionLocal() as session:
        yield session


async def get_write_db() -> AsyncIterator[AsyncSession]:
    """主库会话：成功提交，异常回滚，最后关闭。"""
    async with SessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

async with 负责关闭异步 Session。写依赖中的 commit() 和 rollback() 都必须 await。

2. main.py

from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import Depends, FastAPI
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from database import master_engine, slave_engine
from deps import get_db, get_write_db
from models import Base, User


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 学习阶段按需建表；生产项目使用 Alembic 迁移
    async with master_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    yield

    # 异步引擎必须使用 await dispose()
    await master_engine.dispose()
    if slave_engine is not None:
        await slave_engine.dispose()


app = FastAPI(lifespan=lifespan)


@app.get("/users")
async def list_users(
    db: Annotated[AsyncSession, Depends(get_db)],
):
    result = await db.execute(
        select(User.id, User.name, User.email).order_by(User.id)
    )
    return [dict(row) for row in result.mappings().all()]


@app.post("/users")
async def create_user(
    name: str,
    email: str,
    db: Annotated[AsyncSession, Depends(get_write_db)],
):
    user = User(name=name, email=email)
    db.add(user)          # add() 只是登记对象，不执行网络 I/O，不加 await
    await db.flush()      # 真正执行 INSERT，并取得自增 ID
    return {"id": user.id, "name": user.name}

启动：

uv run uvicorn main:app --reload

Base.metadata.create_all 是同步元数据操作，所以通过 conn.run_sync() 放到异步连接的事务中执行，不能直接写成 await Base.metadata.create_all(...)。

五、异步 Session 的使用规则

操作	写法	是否需要 await
创建会话上下文	async with SessionLocal() as db	使用异步上下文管理器
查询	await db.execute(...)、await db.scalars(...)	需要
主键查询	await db.get(User, user_id)	需要
登记新增对象	db.add(user)	不需要
执行 flush	await db.flush()	需要
删除 ORM 对象	await db.delete(user)	需要
提交、回滚	await db.commit()、await db.rollback()	需要
关闭引擎	await master_engine.dispose()	需要

还需要遵守两条原则：

1. Session 工厂可以全局复用，AsyncSession 不能跨请求共享。 每个请求或任务使用独立 Session。
2. 不要在同一个 AsyncSession 上用 asyncio.gather() 并发执行多个操作。 一个 Session 同时只进行一个数据库工作流；需要并发时使用多个独立 Session，并受连接池容量限制。

总结

1. MySQL 异步连接使用 mysql+aiomysql，引擎使用 create_async_engine()。
2. AsyncEngine 管理异步连接池，async_sessionmaker 创建 AsyncSession 工厂。
3. 数据库 I/O 使用 await，但 db.add() 等纯内存操作不需要 await。
4. 建表使用 await conn.run_sync(Base.metadata.create_all)，关闭引擎使用 await engine.dispose()。
5. 读写分离通过主从会话工厂显式选择，写后读和强一致查询仍走主库。
6. 后续第 21～24 篇统一使用这里的 master_engine、SessionLocal 和 get_db，不再混用同步 PyMySQL 会话。

练习：把 echo 改为 True，分别观察一次查询和一次写入的 SQL；再在 async def 路由中故意漏掉 db.execute() 前的 await，检查结果对象与真正数据库响应的区别。