FastAPI: 数据库集成 — 异步 SQLAlchemy + Alembic

Q: Mapped[str | None] 和 Mapped[Optional[str]] 有区别吗？

功能等价。 str | None 是 Python 3.10+ 语法， Optional[str] 是 typing 兼容写法。推荐前者。

Q: 异步 SQLAlchemy 能用同步代码吗？

不能在 async 函数中直接调用同步 SQLAlchemy 方法。用 run_in_executor 包装或全部使用异步 API。

最后更新：2026-08-26

数据库就像仓库——如果门（连接）开得太少，货物（查询）排队；如果布局（索引）混乱，找一件商品（数据）要翻遍整个仓库。

1. 你将学到

SQLAlchemy 2.0 异步引擎：create_async_engine、AsyncSession
ORM 模型声明：DeclarativeBase、Mapped、mapped_column（2.0 新风格）
Alembic 异步迁移：env.py 配置 run_migrations_online 异步模式
异步会话管理：async with / yield 依赖注入模式
Alice 场景：PriceTracker 的 products/prices/users 三表异步模型设计

2. Alice 的真实故事

(1) 痛点：同步数据库拖慢异步 API

Alice 用同步 SQLAlchemy 操作数据库，每个查询阻塞事件循环 50-100ms。当 PriceTracker 的并发请求达到 500 时，异步 FastAPI 的优势被同步数据库完全抵消，P99 延迟从预期的 50ms 飙到 2000ms。Charlie 说数据库连接池耗尽，新请求在排队等连接。

(2) 异步 SQLAlchemy 的解法

SQLAlchemy 2.0 提供原生异步支持：create_async_engine + AsyncSession，数据库查询不再阻塞事件循环，FastAPI 的异步优势得以完全发挥。

PYTHON

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
async_session = async_sessionmaker(engine, expire_on_commit=False)

async def get_db():
    async with async_session() as session:
        yield session

(3) 收益

迁移到异步 SQLAlchemy 后，PriceTracker 的 P99 延迟从 2000ms 降到 80ms，数据库连接利用率从 30% 提升到 90%，单节点 QPS 从 500 提升到 3000+。

3. 异步引擎与会话

(1) 数据库架构 ER 图

erDiagram
    users ||--o{ products : creates
    products ||--o{ prices : has
    users {
        int id PK
        string email UK
        string hashed_password
        string role
        string subscription
        datetime created_at
    }
    products {
        int id PK
        string name
        string category
        float base_price
        string description
        int user_id FK
        datetime created_at
    }
    prices {
        int id PK
        int product_id FK
        float price
        string currency
        string source
        datetime recorded_at
    }

▶ 示例：异步引擎与连接池配置

PYTHON

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

DATABASE_URL = "postgresql+asyncpg://pricetracker:secret@localhost:5432/pricetracker"

engine = create_async_engine(
    DATABASE_URL,
    echo=False,          # Set True for SQL logging in dev
    pool_size=20,        # Persistent connections
    max_overflow=10,     # Extra connections when pool exhausted
    pool_timeout=30,     # Wait time for available connection
    pool_recycle=3600,   # Recycle connections after 1 hour
)

async_session = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,  # Access objects after commit
)

输出：

TEXT 📖 仅展示

# 执行成功

(2) 同步 vs 异步 SQLAlchemy 对比

维度	同步 SQLAlchemy	异步 SQLAlchemy 2.0
引擎	`create_engine`	`create_async_engine`
会话	`Session`	`AsyncSession`
查询	`session.execute(stmt)`	`await session.execute(stmt)`
提交	`session.commit()`	`await session.commit()`
驱动	psycopg2	asyncpg
阻塞	是	否
连接池	QueuePool	AsyncAdaptedQueuePool

4. ORM 模型声明（2.0 新风格）

(1) DeclarativeBase + Mapped + mapped_column

SQLAlchemy 2.0 用 Mapped[type] 和 mapped_column() 替代旧式的 Column() 声明，类型提示与 ORM 映射统一。

▶ 示例：PriceTracker 三表模型

PYTHON

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Float, Integer, DateTime, ForeignKey, Index
from sqlalchemy.orm import relationship
from datetime import datetime

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    role: Mapped[str] = mapped_column(String(50), default="user")
    subscription: Mapped[str] = mapped_column(String(50), default="free")
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    products: Mapped[list["Product"]] = relationship(back_populates="owner")

class Product(Base):
    __tablename__ = "products"
    __table_args__ = (
        Index("ix_products_category", "category"),
        Index("ix_products_name", "name"),
    )

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(200), nullable=False)
    category: Mapped[str] = mapped_column(String(100), nullable=False)
    base_price: Mapped[float] = mapped_column(Float, nullable=False)
    description: Mapped[str | None] = mapped_column(String(2000), nullable=True)
    user_id: Mapped[int] = mapped_column(Integer, ForeignKey("users.id"))
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    owner: Mapped["User"] = relationship(back_populates="products")
    prices: Mapped[list["Price"]] = relationship(back_populates="product")

class Price(Base):
    __tablename__ = "prices"
    __table_args__ = (
        Index("ix_prices_product_id", "product_id"),
        Index("ix_prices_recorded_at", "recorded_at"),
    )

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    product_id: Mapped[int] = mapped_column(Integer, ForeignKey("products.id"))
    price: Mapped[float] = mapped_column(Float, nullable=False)
    currency: Mapped[str] = mapped_column(String(3), default="USD")
    source: Mapped[str] = mapped_column(String(100))
    recorded_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    product: Mapped["Product"] = relationship(back_populates="prices")

输出：

TEXT 📖 仅展示

# 执行成功

(2) V1 旧风格 vs V2 新风格

维度	V1 旧风格	V2 新风格
Base	`declarative_base()`	`class Base(DeclarativeBase)`
字段声明	`Column(Integer, primary_key=True)`	`Mapped[int] = mapped_column(...)`
类型提示	无	`Mapped[type]` 完整类型
可选字段	`Column(String, nullable=True)`	`Mapped[str \| None]`
关系	`relationship()`	`Mapped[list["X"]] = relationship()`

5. 异步会话管理与依赖注入

(1) yield 依赖模式

▶ 示例：DB Session 依赖

PYTHON

from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

输出：

TEXT 📖 仅展示

# 函数定义成功

▶ 示例：在端点中使用异步会话

PYTHON

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

app = FastAPI()

@app.get("/products/{product_id}")
async def get_product(
    product_id: int,
    db: AsyncSession = Depends(get_db),
):
    stmt = select(Product).where(Product.id == product_id)
    result = await db.execute(stmt)
    product = result.scalar_one_or_none()
    if not product:
        from fastapi import HTTPException
        raise HTTPException(status_code=404, detail="Product not found")
    return {
        "id": product.id,
        "name": product.name,
        "base_price": product.base_price,
    }

输出：

TEXT 📖 仅展示

# 函数定义成功

6. Alembic 异步迁移

(1) 迁移工作流

flowchart LR
    A[alembic revision --autogenerate -m desc] --> B[Edit Migration File]
    B --> C[alembic upgrade head]
    C --> D[Apply to Database]
    D --> E{Need Rollback?}
    E -->|Yes| F[alembic downgrade -1]
    E -->|No| G[Continue Development]

▶ 示例：初始化 Alembic

BASH

# Install Alembic
uv add alembic

# Initialize Alembic with async template
cd pricetracker
alembic init -t async alembic

输出：

TEXT 📖 仅展示

# 命令执行成功

▶ 示例：配置 alembic/env.py 异步模式

PYTHON

# alembic/env.py (key sections)
from sqlalchemy.ext.asyncio import create_async_engine
from app.models import Base  # Import your models
from app.core.config import settings

target_metadata = Base.metadata

def run_migrations_online():
    connectable = create_async_engine(settings.database_url)

    async def do_run_migrations(connection):
        context = MigrationContext.configure(
            connection=connection,
            target_metadata=target_metadata,
        )
        with context.begin_transaction():
            context.run_migrations()

    with connectable.connect() as connection:
        asyncio.run(do_run_migrations(connection))

输出：

TEXT 📖 仅展示

# 函数定义成功

▶ 示例：生成与应用迁移

BASH

# Auto-generate migration from model changes
alembic revision --autogenerate -m "add users products prices tables"

# Apply migration
alembic upgrade head

# Rollback one step
alembic downgrade -1

# Check current version
alembic current

输出：

TEXT 📖 仅展示

# 命令执行成功

❓ 常见问题

Q asyncpg 和 psycopg 有什么区别？

A asyncpg 是纯异步 PostgreSQL 驱动，性能更好；psycopg3 支持异步但基于 C 库。生产推荐 asyncpg（postgresql+asyncpg://）。

Q expire_on_commit=False 什么意思？

A 默认 commit 后 ORM 对象属性过期（再访问触发查询）。设为 False 保持属性可访问，避免延迟加载问题。

Q Alembic 的 autogenerate 能检测所有变更吗？

A 不能。能检测表增删、列增删、索引变更。不能检测列名重命名、约束语义变更等，需手动编辑迁移文件。

Q 连接池大小怎么设？

A 公式：pool_size = (CPU cores * 2) + 有效磁盘数。PriceTracker 用 pool_size=20 + max_overflow=10 应对百万级查询。

Q Mapped[str | None] 和 Mapped[Optional[str]] 有区别吗？

A 功能等价。str | None 是 Python 3.10+ 语法，Optional[str] 是 typing 兼容写法。推荐前者。

Q 异步 SQLAlchemy 能用同步代码吗？

A 不能在 async 函数中直接调用同步 SQLAlchemy 方法。用 run_in_executor 包装或全部使用异步 API。

📖 小节

SQLAlchemy 2.0 异步引擎（create_async_engine + AsyncSession）不阻塞事件循环，充分发挥 FastAPI 异步优势
Mapped[type] + mapped_column() 是 2.0 新风格，类型提示与 ORM 映射统一
yield 依赖模式管理异步会话生命周期：自动 commit/rollback/close
Alembic 异步迁移用 -t async 模板初始化，env.py 配置异步引擎
PriceTracker 三表模型（users/products/prices）含索引策略，支持百万级数据查询

📝 作业

基础题（难度⭐）：配置 create_async_engine 连接 PostgreSQL，创建 async_sessionmaker，编写 get_db yield 依赖。提示：create_async_engine(DATABASE_URL, pool_size=20)
进阶题（难度⭐⭐）：定义 PriceTracker 的 Product 和 Price 模型（DeclarativeBase + Mapped 风格），包含外键关系和索引。提示：ForeignKey("products.id") + relationship()
挑战题（难度⭐⭐⭐）：初始化 Alembic 异步模式，配置 env.py，用 autogenerate 生成三表迁移文件并 upgrade head 应用到数据库。提示：alembic init -t async alembic + 修改 env.py 的 target_metadata

---|

Web-Tutorial 技术团队

由多位开发者共同维护的编程教程平台。每篇教程由对应领域的开发者编写和审核，确保内容准确可靠。如发现任何问题，欢迎向我们反馈。

关于我们

FastAPI: 数据库集成 — 异步 SQLAlchemy + Alembic

1. 你将学到

2. Alice 的真实故事

(1) 痛点：同步数据库拖慢异步 API

(2) 异步 SQLAlchemy 的解法

(3) 收益

3. 异步引擎与会话

(1) 数据库架构 ER 图

▶ 示例：异步引擎与连接池配置

(2) 同步 vs 异步 SQLAlchemy 对比

4. ORM 模型声明（2.0 新风格）

(1) DeclarativeBase + Mapped + mapped_column

▶ 示例：PriceTracker 三表模型

(2) V1 旧风格 vs V2 新风格

5. 异步会话管理与依赖注入

(1) yield 依赖模式

▶ 示例：DB Session 依赖

▶ 示例：在端点中使用异步会话

6. Alembic 异步迁移

(1) 迁移工作流

▶ 示例：初始化 Alembic

▶ 示例：配置 alembic/env.py 异步模式

▶ 示例：生成与应用迁移

❓ 常见问题

📖 小节

📝 作业

🙏 帮我们做得更好