usehook
创建时间: 2026-09-01最后更新: 2026-09-01

1. 认识 SQLAlchemy

Psycopg 已经可以完成所有数据库操作，但项目规模扩大以后，如果每个函数都要手动处理连接、游标、结果转换和重复 SQL，维护成本会逐渐增加。SQLAlchemy 建立在数据库驱动之上，帮助应用统一组织 SQL、连接、事务和对象映射。

SQLAlchemy 在驱动之上提供两套相关能力：

能力作用
Core使用 Python 对象构造 SQL 表达式，管理 Engine、连接和事务
ORM把表映射为 Python 类，在 Core 之上管理对象状态和关系

Core 和 ORM 不是两套互不相关的技术。ORM 建立在 Core 之上，最终仍然把 Python 操作转换成 SQL，再通过 Psycopg 发送给 PostgreSQL。它不会取代数据库约束，也不会让我们可以忽略 SQL 和执行计划。

同一个项目也不必在 Core 和 ORM 之间二选一。常规的增删改查可以使用 ORM 组织对象，批量更新、复杂统计或 PostgreSQL 专用语句则可以直接使用 Core 表达式。选择依据是查询是否清楚、可维护，而不是强制所有 SQL 都写成同一种形式。

sqlalchemy-layers.txt
1
FastAPI 业务代码

2
  -> SQLAlchemy ORM

3
  -> SQLAlchemy Core / Engine / Pool / Dialect

4
  -> Psycopg

5
  -> PostgreSQL

安装 SQLAlchemy 与 Psycopg：

install-sqlalchemy.bash
1
uv add "sqlalchemy[asyncio]>=2.0,<3" "psycopg[binary]>=3.2"

asyncio 额外项会确保 greenlet 等 SQLAlchemy 异步功能需要的依赖可用。接下来使用 DeclarativeBase 创建声明基类，并通过 Mapped 和 mapped_column() 定义类型化模型。

对于刚接触 SQLAlchemy 的读者，可以先记住接下来几篇文章的分工：Engine 负责连接数据库，声明式模型描述表与 Python 类的对应关系，Session 则负责一次业务操作中的查询、对象状态和事务。本篇先完成前两部分，下一篇再进入 Session。

2. Engine

Engine 是应用访问数据库的入口，它包含方言和连接池。异步 FastAPI 项目使用 create_async_engine()：

app/db/engine.py
1
from sqlalchemy.ext.asyncio import create_async_engine

2


3
from app.core.settings import settings

4


5


6
engine = create_async_engine(

7
    settings.database_url,

8
    pool_pre_ping=True,

9
)

配置中的 URL 使用 SQLAlchemy 方言格式：

.env.example
1
DATABASE_URL=postgresql+psycopg://task_app:task_app_dev_password@127.0.0.1:5432/task_app

这段 URL 可以拆成下面几部分：

组成部分示例值作用
数据库方言postgresql告诉 SQLAlchemy 使用 PostgreSQL 语法和类型
数据库驱动psycopg选择实际负责通信的 Python 驱动
用户名与密码task_app:task_app_dev_password用于数据库身份认证
主机与端口127.0.0.1:5432指定 PostgreSQL 服务地址
数据库名task_app指定连接的目标数据库

postgresql 选择 PostgreSQL 方言，psycopg 选择驱动。与 create_async_engine() 组合后，SQLAlchemy 会使用 Psycopg 的异步实现。同一个 postgresql+psycopg:// 方言也支持同步 Engine，究竟使用同步还是异步实现，由 create_engine() 或 create_async_engine() 决定。

如果用户名或密码中含有 @、:、/ 等 URL 特殊字符，需要进行百分号编码，不能直接拼进 URL。应用代码也不应把完整数据库 URL 写入日志，因为其中通常包含密码。

这里可以把 Engine 的职责拆成三个部分理解：

  • Dialect 知道 PostgreSQL 的 SQL 语法、类型和驱动调用方式；
  • Pool 复用已经建立的数据库连接；
  • Engine 统一提供连接、事务和 SQL 执行入口。

这里创建出来的对象是 AsyncEngine。它通常是进程级长生命周期对象，不应该在每个请求中重复创建。调用 create_async_engine() 时一般也不会立刻连接数据库，第一次真正取得连接或执行 SQL 时，连接池才会建立连接。因此，仅仅导入模块或启动应用没有报错，并不能证明数据库 URL 一定可用。

可以执行一条轻量查询检查连接：

check-database.py
1
from sqlalchemy import text

2


3
from app.db.engine import engine

4


5


6
async def check_database() -> bool:

7
    async with engine.connect() as connection:

8
        value = await connection.scalar(text("SELECT 1"))

9
        return value == 1

engine.connect() 从连接池取得一条连接，离开上下文时再归还。SQLAlchemy 2.x 默认会在第一次执行语句时自动开始事务；这里是只读检查，连接归还前会结束事务。需要提交写入时，应使用 engine.begin()、显式事务或下一篇介绍的 Session 事务，不能把关闭连接误解为自动提交。

pool_pre_ping=True 会在连接池借出连接时检查它是否仍然可用，减少数据库重启或网络断开后拿到失效连接的问题。它会增加一次轻量检查，但不能挽救执行到一半时断开的事务；这种情况仍会抛出异常，应用必须回滚，并根据操作的幂等性决定是否重试整个事务。

应用关闭时应执行 await engine.dispose()，让连接池主动关闭空闲连接。这个操作属于 FastAPI lifespan 的关闭阶段，而不是每个请求的清理步骤：

app/main.py
01
from collections.abc import AsyncIterator

02
from contextlib import asynccontextmanager

03


04
from fastapi import FastAPI

05


06
from app.db.engine import engine

07


08


09
@asynccontextmanager

10
async def lifespan(_app: FastAPI) -> AsyncIterator[None]:

11
    try:

12
        yield

13
    finally:

14
        await engine.dispose()

15


16


17
app = FastAPI(lifespan=lifespan)

连接池容量、worker 数量和 PostgreSQL 的连接上限要一起规划，不能把每个进程的池大小孤立看待。这里暂时使用默认池配置，生产环境中的 pool_size、max_overflow 和超时设置会在部署章节单独讨论。

正在验证登录状态
请稍候，验证完成后将继续显示文章内容
0
订阅