连上数据库：SQLAlchemy 与 Engine

本文目录

API 写到一半，Postman 里 JSON 已经漂亮，一持久化才发现：连接串散落在三个文件、本地 SQLite 与 Docker 里的 Postgres 切换要改代码、测试库和开发库不小心指到同一个 URL。SQLAlchemy 的第一课不是写查询，而是把 Engine 和连接配置收拢到一处——否则后面 ORM、迁移、事务全建在沙地上。

Engine 是什么

SQLAlchemy 里 Engine 是到数据库的工厂 + 连接池入口。你很少「持有一个长连」写全程；而是向 Engine 要连接，用完归还池子。

直觉分层：

对象	职责
Engine	解析 URL、管理池、方言（Postgres/SQLite/MySQL）
Connection	一次 DBAPI 连接上的操作上下文
Session（下篇）	ORM 级工作单元，挂在 Engine 上

FastAPI 本身不管数据库；通常在 lifespan 或依赖注入里创建 Engine，请求结束时释放连接。

URL 与 create_engine

数据库 URL 遵循 RFC 风格：

dialect+driver://username:password@host:port/database

示例：

python

# shop/db/engine.py
from sqlalchemy import create_engine
from sqlalchemy.engine import Engine

from shop.core.config import settings


def build_sync_engine() -> Engine:
    return create_engine(
        settings.DATABASE_URL,
        pool_pre_ping=True,      # 取连接前 ping，减少 stale connection
        pool_size=5,
        max_overflow=10,
        echo=settings.SQL_ECHO,  # 开发时打印 SQL
    )


engine: Engine = build_sync_engine()

create_engine 不会立刻连数据库（lazy）；第一次要连接时才建立池。pool_pre_ping 在生产环境很常见，避免中间网络设备踢 idle 连接后第一次 query 报错。

SQLite 本地开发：

python

DATABASE_URL = "sqlite:///./dev.db"
# 内存库：sqlite:///:memory:

Postgres：

python

DATABASE_URL = "postgresql+psycopg://user:pass@localhost:5432/shop"

驱动名要与你安装的 DBAPI 包一致（psycopg v3、psycopg2、asyncpg 等）。

异步：create_async_engine

FastAPI 路由多为 async def；若数据库驱动支持 asyncio，可用异步 Engine：

python

from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine

from shop.core.config import settings


def build_async_engine() -> AsyncEngine:
    return create_async_engine(
        settings.ASYNC_DATABASE_URL,
        pool_pre_ping=True,
    )


# postgresql+asyncpg://user:pass@localhost:5432/shop
async_engine = build_async_engine()

注意：同步 URL postgresql+psycopg://... 与异步 postgresql+asyncpg://... 不是简单互换——驱动不同。团队应选定一条路径（全 async 或 ORM 同步 + 线程池），避免混用两套 Engine 无章法。

验证连接是否通：

python

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncEngine


async def check_connection(engine: AsyncEngine) -> bool:
    async with engine.connect() as conn:
        result = await conn.execute(text("SELECT 1"))
        return result.scalar_one() == 1

text() 包裹原生 SQL 字符串；connect() 上下文退出时连接归还池，不是关整个 Engine。

配置抽取：别硬编码 URL

配置应来自环境变量或 .env（不提交密钥到 git）：

python

# shop/core/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

    PROJECT_NAME: str = "Shop API"
    API_V1_PREFIX: str = "/api/v1"

    DATABASE_URL: str = "sqlite:///./dev.db"
    ASYNC_DATABASE_URL: str = "sqlite+aiosqlite:///./dev.db"
    SQL_ECHO: bool = False


settings = Settings()

.env 示例：

DATABASE_URL=postgresql+psycopg://shop:secret@db:5432/shop
ASYNC_DATABASE_URL=postgresql+asyncpg://shop:secret@db:5432/shop
SQL_ECHO=false

测试环境在 CI 里 export 另一套 URL，或在 conftest.py 覆盖 Settings，避免误删开发数据。

与 FastAPI lifespan 挂钩

Engine 生命周期应跟应用进程一致：

python

# shop/main.py 片段
from contextlib import asynccontextmanager

from fastapi import FastAPI

from shop.db.engine import async_engine


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 可选：启动时 check_connection
    yield
    await async_engine.dispose()  # 进程退出前释放池


app = FastAPI(lifespan=lifespan)

忘记 dispose() 在测试套件里可能留下 hanging 连接，导致「跑完 pytest 进程不退」。

执行层：Engine 之上还有什么

本篇止于 Engine 与连接。实际 CRUD 会走 Session（同步）或 AsyncSession（异步）——它们在内部从 Engine 借连接。raw SQL 也可 engine.connect() + conn.execute()，适合迁移脚本或极简单查询；业务 ORM 仍推荐 Session 边界（第 5 篇）。

配置里还可扩展：pool_recycle（MySQL 常见）、connect_args（SQLite 开 WAL）、读写分离多个 Engine——按部署规模加，小项目默认池参数够用。

连接池参数怎么理解

pool_size 是池里常驻连接数；max_overflow 允许在高峰临时多借的连接，用完释放。设太大，数据库端 max_connections 可能被吃满；设太小，高并发下请求排队等连接。开发环境 pool_size=5 往往够用；生产按 QPS 与平均查询耗时估算，并用监控看「等待池超时」指标。

echo=True 时 SQLAlchemy 把 SQL 打到日志，排查 N+1 或慢查询很方便，但生产别长期开——日志量与敏感参数泄露风险都高。用配置开关 SQL_ECHO，本地 true、线上 false。

SQLite 特殊：check_same_thread=False 有时要写在 connect_args 里配合多线程 Uvicorn worker；更好的做法仍是「每 worker 一个 Engine / 池」，而不是跨线程共享同一连接。

开发库与生产库：同一套配置代码

典型做法是代码只认 DATABASE_URL，三份 env 文件指向不同实例：

环境	文件	URL 指向
本地	`.env`	SQLite 或 Docker Postgres
CI	pipeline env	临时 Postgres 容器
生产	密钥管理 / K8s Secret	托管 RDS

配置类不要写 if os.getenv("ENV") == "prod" 散落分支；用 pydantic-settings 的 env_file 或部署层注入即可。测试误连生产是事故级问题——CI 里显式 export DATABASE_URL=postgresql://test:test@localhost/test，并在配置校验里禁止 test 进程使用含 prod 字样的 host（可选硬 guard）。

健康检查与启动探针

Kubernetes 存活探针可打纯 FastAPI 路由；就绪探针最好包含数据库连接探测：

python

from fastapi import APIRouter, HTTPException
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncEngine

router = APIRouter()


@router.get("/health/db")
async def health_db(engine: AsyncEngine) -> dict[str, str]:
    try:
        async with engine.connect() as conn:
            await conn.execute(text("SELECT 1"))
    except Exception as exc:
        raise HTTPException(status_code=503, detail="db unavailable") from exc
    return {"database": "ok"}

启动时 lifespan 里做一次 check_connection 能更早发现 URL 配错；运行时探针则捕获 DB 中途宕机。两者互补，不是二选一。

Engine 与 Session 的分工预告

初学者常问：既然有 Engine，为什么还要有 Session？Engine 管连接池与方言；Session 管 ORM 对象跟踪与事务边界。你可以在 Engine 上直接 connect() 执行 SQL，但一涉及 Model 实例的新增、修改、关系加载，Session 省大量样板代码。下一篇定义映射；第 5 篇在 Session 里做 CRUD。记住：Engine 是进程级长活对象；Session 是请求级或短事务级。

方言与驱动：URL 里藏着的细节

postgresql+psycopg:// 里的 psycopg 是 DBAPI 驱动；换成 psycopg2 要改 URL 与依赖包。MySQL 常见 mysql+pymysql://。SQLite 文件路径用三个斜杠 sqlite:///./file.db；相对路径相对当前工作目录，Docker 里工作目录不对会导致「库文件写到意外位置」。配置里用绝对路径或挂载卷路径更稳。

create_engine 的 connect_args 可传驱动专属参数，例如 SQLite 的 timeout=30 缓解 locked。SQLAlchemy 负责把 Python 调用翻译成带方言的 SQL；换数据库时改 URL 与少量类型（如 JSONB vs JSON），Engine 接口不变。

本地 Docker Compose 里数据库 hostname 常是服务名 db，而本机直连用 localhost——同一份镜像、两份配置 env 即可，代码零分支。CI 里用 ephemeral 容器时，迁移与 Engine 指向同一 URL，避免测的是 A 库、跑的是 B 库。

Secrets 不要写进代码仓库：DATABASE_URL 从环境注入，日志里打印连接串前要把密码打码。Settings 类可以加 validator，在 production 检测到空密码或默认 secret 时启动失败——把配置错误拦在 import 阶段，而不是第一条查询才爆。

线程池里跑同步 Engine 时，每个线程仍从同一池借连接；Uvicorn 多 worker 则是多进程、每进程一池。调 pool_size 时要按 worker 数乘起来估算总连接数，避免把 Postgres max_connections 撑爆。

观测连接池：日志里若出现 QueuePool limit ... timeout，说明等在池外的查询过多——要么加大池（短期），要么减慢查询、加缓存、或读 replica（长期）。Engine 不是无限水龙头；配置里的数字要和 DBA 配额对齐。

测试里常用 SQLite 内存库 URL：sqlite:///:memory: 或 sqlite+aiosqlite:///:memory:。与 Postgres 方言仍有差异（JSON 函数、部分 DDL），集成测试关键路径仍建议 Docker Postgres；单元测试用 SQLite 换速度。配置通过 env 切换，不在代码里写 if testing。

future=True 在 2.0 已是默认；老教程里的 engine.execute("SELECT 1") 应改为 with engine.connect() as conn: conn.execute(text(...))。Engine API 变干净后，连接与事务边界更显式——习惯用上下文管理器，泄漏更少。

把 Engine 放进 Depends 一般不划算——Depends 适合请求级 Session；Engine 在 lifespan 创建一次即可。若测试要换内存库，override 的是 get_db 或 Settings 里的 URL，而不是每次请求 create_engine。

读写分离时两个 Engine 两个 URL：read_engine 与 write_engine 池参数可不同——只读 replica 可开大池。服务层显式传 use_replica=True 比在 Model 上魔法切换更可测；配置里声明两个 URL，文档写清哪些查询允许滞后。

常见误解

误解一：「create_engine 等于已经连上 DB。」 只是创建池；网络/auth 错误可能第一次 query 才暴露。

误解二：「async 路由必须 async Engine。」 可以用同步 Session + run_in_threadpool 包一层，但高并发下 async 驱动更省线程。

误解三：「SQLAlchemy 2.0 还要 sessionmaker(bind=engine) 老写法。」 2.0 风格推荐 sessionmaker 配合 Session 上下文，或 async_sessionmaker；写法变但 Engine 角色不变。

误解四：「把 URL 写在 docker-compose 里就够了。」 应用配置仍应读环境变量，镜像同一套，换环境只换 env。

误解五：「一个 Engine 只能连一个库。」 只连一个实例是常态；读写分离时才建两个 Engine，分别挂不同 URL，服务层决定读哪个写哪个。

误解六：「dispose 只在进程退出才需要。」 测试 session 结束、热重载旧进程退出时也应 dispose，否则连接计数泄漏。

小结

SQLAlchemy 的 Engine 统一管理方言、池与连接；用配置类集中 DATABASE_URL，启动时 create_engine / create_async_engine，关闭时 dispose()。牢记 Engine 进程级、Session 请求级，池参数与 worker 数相乘才是数据库看到的连接总数。先保证连接可测、可配、可关，再写 ORM。下一篇把表映射成 ORM Model，对象与行如何对应。

Engine 是什么 ​

URL 与 create_engine ​

异步：create_async_engine ​

配置抽取：别硬编码 URL ​

与 FastAPI lifespan 挂钩 ​

执行层：Engine 之上还有什么 ​

连接池参数怎么理解 ​

开发库与生产库：同一套配置代码 ​

健康检查与启动探针 ​

Engine 与 Session 的分工预告 ​

方言与驱动：URL 里藏着的细节 ​

常见误解 ​

小结 ​

Engine 是什么

URL 与 create_engine

异步：create_async_engine

配置抽取：别硬编码 URL

与 FastAPI lifespan 挂钩

执行层：Engine 之上还有什么

连接池参数怎么理解

开发库与生产库：同一套配置代码

健康检查与启动探针

Engine 与 Session 的分工预告

方言与驱动：URL 里藏着的细节

常见误解

小结