sqlalchemy

sqlalchemy — SKILL.md

变体： 标准 · 适用时机： 调用此技能后，可构建出可运行的多方言 SQLAlchemy 数据层（或其相关部分），然后将控制权交还给调用方。

概述

此技能指导 Agent 使用 SQLAlchemy 2.x 构建可移植的关系型数据层——一套代码库，其中的模型、会话和查询无需修改即可运行于 SQLite、PostgreSQL 和 MySQL。它以类型化的 2.x ORM（DeclarativeBase、Mapped[...]、mapped_column(...)）为主，仅在 ORM 无法以可移植方式表达少数功能（特定方言的 upsert、RETURNING）时才使用 Core。它优先采用同步方式：主要演示路径为同步，异步仅作简短补充。核心内容是跨方言问题矩阵——少数会因引擎而异、且容易被忽略的行为（行锁、JSON、upsert、标识列、隔离级别）；如果不针对这些行为进行分支处理，就会破坏“一次编写，在三个数据库上运行”的假设。

何时启用

• ✅ 为目标数据库为 SQLite、PostgreSQL 和／或 MySQL 的应用构建 SQLAlchemy Engine、类型化 DeclarativeBase／Mapped 模型，以及会话范围的事务。
• ✅ 为某个方言选择 DBAPI 驱动程序（同步或异步），并编写连接 URL。
• ✅ 为可移植数据层配置连接池、事务范围或隔离级别。
• ✅ 遇到“在 SQLite 上可用，在 Postgres 上失败”（或反之）的情况，例如行锁、JSON、upsert、自增、隔离级别问题。

以下情况请勿启用：

• 编写／运行架构迁移或执行 ALTER TABLE 变更 → alembic。
• 构建作业队列／调度器——就绪集查询、原子租约循环、心跳、公平份额 → sql-job-queue。（此技能仅介绍 with_for_update(skip_locked=...) 这一基本操作。）
• 使用 Pydantic 验证／序列化数据 → pydantic-v2。
• 使用其他 ORM（Django ORM、Tortoise、SQLModel）或原始 DBAPI。

工作流

第 1 步：选择方言和驱动程序，编写 URL

URL 格式为 dialect+driver://user:password@host:port/dbname。按方言选择默认驱动程序（见第 5 步的矩阵）；驱动程序对应 +driver 部分。

# SQLite — file (the +pysqlite / sqlite3 stdlib driver is the default)
"sqlite+pysqlite:///./app.db"          # relative file
"sqlite:///./app.db"                    # same thing — sqlite3 is the default driver
"sqlite:////absolute/path/app.db"       # note the 4th slash = absolute path
"sqlite://"                             # in-memory (also sqlite:///:memory:)

# PostgreSQL — psycopg (v3, the modern maintained driver)
"postgresql+psycopg://user:pw@localhost:5432/appdb"

# MySQL / MariaDB — PyMySQL (pure-Python, no C build step)
"mysql+pymysql://user:pw@localhost:3306/appdb"

第 2 步：每个进程创建一个 Engine

Engine 管理连接池。在进程启动时创建一次，并在各处共享；不要为每个请求／任务都创建一个 Engine。

from sqlalchemy import create_engine

engine = create_engine(
    "postgresql+psycopg://user:pw@localhost/appdb",
    pool_pre_ping=True,      # recycle dead connections (server DBs — PG/MySQL)
    pool_size=5,             # persistent pooled connections
    max_overflow=10,         # extra burst connections beyond pool_size
)

不同方言的连接池默认值有所不同——请参阅“注意事项”。对于服务器数据库（PG／MySQL），请保留 pool_pre_ping=True，这样可以检测到被服务器空闲超时机制关闭的连接，并替换它，而不是等到查询执行中途才报错。

第 3 步：定义类型化的 2.x ORM 模型

继承 DeclarativeBase；使用 Mapped[...] 注解列，并通过 mapped_column(...) 配置。切勿使用旧式的 Column = ...／declarative_base() 形式。

from datetime import datetime
from sqlalchemy import JSON, ForeignKey, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship

class Base(DeclarativeBase):
    pass

class Job(Base):
    __tablename__ = "jobs"

    id: Mapped[int] = mapped_column(primary_key=True)          # autoincrements on all 3
    name: Mapped[str] = mapped_column(index=True)              # NOT NULL (no Optional)
    data: Mapped[dict] = mapped_column(JSON)                   # portable JSON (Step 7)
    note: Mapped[str | None]                                   # NULLable (Optional -> nullable)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())

    runs: Mapped[list["Run"]] = relationship(back_populates="job")

class Run(Base):
    __tablename__ = "runs"
    id: Mapped[int] = mapped_column(primary_key=True)
    job_id: Mapped[int] = mapped_column(ForeignKey("jobs.id"))
    job: Mapped[Job] = relationship(back_populates="runs")

单独使用 Mapped[str] 表示 NOT NULL；Mapped[str | None] 表示可空——SQLAlchemy 会根据注解推导可空性。整数主键会在所有方言中自动递增（见第 9 步）。

第 4 步：初始化架构（仅用于开发／测试），然后创建会话并查询

Base.metadata.create_all(engine)   # DEV/TEST bootstrap ONLY — see Anti-patterns

对于实际项目中任何受管理的架构变更，请使用 alembic；不要将 create_all() 作为迁移方案发布。

每个工作单元使用一个短生命周期会话，并显式开启事务：

from sqlalchemy import select
from sqlalchemy.orm import Session, sessionmaker

SessionLocal = sessionmaker(engine)   # a factory; bind the engine once

# Per unit of work: open, transact, close.
with Session(engine) as session, session.begin():   # begin() = commit on exit, rollback on error
    session.add(Job(name="ingest", data={"src": "s3"}))
    # block exit commits automatically

# Reads
with Session(engine) as session:
    job = session.get(Job, 1)                                    # typed Job | None by PK
    rows = session.execute(
        select(Job).where(Job.name == "ingest").order_by(Job.id)
    ).scalars().all()                                            # list[Job]

session.execute(select(...)) 会返回 Row 元组；调用 .scalars() 可提取其中的 ORM 实体。session.get(Model, pk) 是按主键查询的快捷路径（会先检查标识映射）。

步骤 5：选择同步或异步模式及驱动（对照表）

同步是首选路径。 只有当周边技术栈已经是异步的（例如 ASGI 应用）时，才选择异步模式。URL 中的 +driver 用于选择同步或异步模式——没有对应的标志。

方言	默认同步驱动	URL 前缀	默认异步驱动	异步 URL 前缀	备选方案（单行说明理由）
SQLite	sqlite3（pysqlite，标准库）	sqlite+pysqlite:// / sqlite://	aiosqlite	sqlite+aiosqlite://	—（标准库驱动是唯一主流的同步选择）
PostgreSQL	psycopg（v3）	postgresql+psycopg://	asyncpg	postgresql+asyncpg://	psycopg2（旧版、成熟、仅支持同步——仅用于现有的 v2 部署）；如果希望一个驱动同时支持两种模式，psycopg v3 也可用于异步模式
MySQL / MariaDB	PyMySQL	mysql+pymysql://	aiomysql	mysql+aiomysql://	mysqlclient（C 扩展，速度更快，需要构建工具链）；asyncmy（异步速度更快，采用 Cython 构建）

选择这些默认项的原则是：维护良好且使用门槛最低——psycopg v3 是现代且积极维护的 PG 驱动；PyMySQL 采用纯 Python 实现，因此安装时无需编译器；asyncpg 是速度最快的异步 PG 驱动。

步骤 6：异步补充说明（简短）

模型类相同；换用引擎和会话系列，并对 I/O 使用 await。

from sqlalchemy import select
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

aengine = create_async_engine("postgresql+asyncpg://user:pw@localhost/appdb")
ASessionLocal = async_sessionmaker(aengine, expire_on_commit=False)

async with ASessionLocal() as session:
    async with session.begin():
        session.add(Job(name="async-ingest", data={}))
    res = await session.execute(select(Job))
    jobs = res.scalars().all()

异步模式通过 greenlet 桥接来执行同步 DBAPI 操作——你编写 await，SQLAlchemy 会处理切换。需要注意的一点是：提交后，访问属性可能会触发惰性加载，而这在异步模式下会引发异常（不允许隐式 I/O）。设置 expire_on_commit=False，和/或对提交后要访问的关联进行预加载（selectinload）。切勿在并发任务间共享同一个 AsyncSession（与在线程间共享同步会话时遵循的规则相同）。

第 7 步：跨方言注意事项——针对差异分支处理

这是至关重要的一节。每项内容的完整示例请参见 references/sqlalchemy-extras.md。可移植规则如下：

行锁——支持矩阵（至关重要）。 select(...).with_for_update(...) 会生成 SELECT ... FOR UPDATE。

stmt = select(Job).where(Job.name == "ingest").with_for_update(
    skip_locked=True,   # skip rows another txn holds (claim-next pattern)
    nowait=False,       # True = error instead of waiting if a row is locked
    read=False,         # True = FOR SHARE instead of FOR UPDATE
    of=Job,             # lock only this table's rows in a multi-table select
)

方言	FOR UPDATE	SKIP LOCKED / NOWAIT
PostgreSQL	支持	支持
MySQL	支持（InnoDB）	支持——仅限 MySQL 8.0+
SQLite	不支持——没有行级锁（使用文件／数据库级锁）；with_for_update 实际上是空操作

必须在 SQLite 上运行的数据层不能依赖行锁。SQLite 上的替代方案（BEGIN IMMEDIATE 声明策略）属于同系列 sql-job-queue 的范畴——请转到该项目，不要在这里实现。

JSON。 为保证可移植性，请使用通用的 sqlalchemy.JSON——在 PG/MySQL 上存储为原生 JSON，在 SQLite 上存储为 TEXT（通过 SQLite 的 JSON1）。可移植子集是存取完整文档。如需使用 PostgreSQL 原生索引／包含查询，请改用 sqlalchemy.dialects.postgresql.JSONB——但它仅适用于 PG。各方言的路径／包含查询运算符有所不同；不要假设为某个引擎编写的 JSON WHERE 子句能在其他引擎上运行。

Upsert——没有可移植形式；按方言分支处理。 从方言模块导入 insert：

# PostgreSQL (and SQLite — same API, different import)
from sqlalchemy.dialects.postgresql import insert   # or .sqlite import insert
stmt = insert(Job).values(id=1, name="ingest", data={})
stmt = stmt.on_conflict_do_update(index_elements=["id"], set_={"data": stmt.excluded.data})
# stmt.on_conflict_do_nothing(index_elements=["id"])   # the do-nothing variant
# index_elements must name a real UNIQUE / PK constraint (or partial-index cols), else no conflict is detected

# MySQL
from sqlalchemy.dialects.mysql import insert
stmt = insert(Job).values(id=1, name="ingest", data={})
stmt = stmt.on_duplicate_key_update(data=stmt.inserted.data)

注意，“将要插入的行”在 PG/SQLite 中使用 excluded，在 MySQL 中使用 inserted。

自增／标识列。 整数主键（Mapped[int] = mapped_column(primary_key=True)）在这三种方言中都会自增——PG 使用 IDENTITY/SERIAL，MySQL 使用 AUTO_INCREMENT，SQLite 使用 ROWID。默认的可移植方案即可正常工作；不要手动编写序列逻辑。

事务隔离。 可按引擎或连接设置：

engine = create_engine(url, isolation_level="REPEATABLE READ")
# or per-connection / per-block:
with engine.connect().execution_options(isolation_level="SERIALIZABLE") as conn:
    ...

有效级别：READ UNCOMMITTED、READ COMMITTED、REPEATABLE READ、SERIALIZABLE、AUTOCOMMIT。默认值有所不同：PostgreSQL = READ COMMITTED，MySQL/InnoDB = REPEATABLE READ。在 SQLite 上，SQLAlchemy 内置支持的可设置 isolation_level 只有 AUTOCOMMIT（引擎在其他情况下近似可串行化）——不要指望 isolation_level="SERIALIZABLE" 能在那里生效。如果正确性依赖某个隔离级别，请明确设置——不要沿用方言默认值。

规则

硬性规则（绝不违反）：

• 仅使用 2.x 类型化 ORM。 使用 DeclarativeBase + Mapped[...] + mapped_column(...)。绝不使用旧式 Column = / declarative_base() 形式。
• 每个进程仅使用一个引擎。 引擎拥有连接池；创建一次，处处共享。
• Session 生命周期短且不共享。 每个工作单元使用一个 Session/AsyncSession；绝不在线程或异步任务之间共享 session。
• create_all() 仅用于开发／测试环境的引导。 受管理的架构变更应使用 alembic。
• 针对跨方言注意事项分支处理。 绝不假设行锁、单一的 upsert 形式、JSON 查询运算符或隔离级别默认值可以在 SQLite/PG/MySQL 之间通用。

偏好（可覆盖）：

• 除非有明确约束（例如现有 psycopg2 部署、对性能至关重要的异步路径）证明替代方案合理，否则优先使用各方言的默认驱动（第 5 步）。
• 优先使用 ORM；仅当 ORM 无法可移植地表达所需操作时，才改用 Core（方言 upsert、RETURNING）。
• 对 PG/MySQL 保留 pool_pre_ping=True；SQLite 无需设置。
• 优先使用同步方式；仅在已有的异步技术栈中采用异步方式。

易踩坑

• sqlite:///path 中斜杠的数量。 三个斜杠 + 相对路径表示相对于 CWD；绝对路径需要四个（sqlite:////abs/path.db）。内存数据库使用 sqlite:// 或 sqlite:///:memory:。
• 新建的 sqlite:// 每个连接都是空的。 普通内存引擎会为每个连接池连接创建其自己的数据库——在一个连接上创建的表对下一个连接不可见。要使用共享内存数据库（用于测试），请使用 StaticPool（一个共享连接）：create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)。
• 不同方言的连接池默认值不同。 文件 SQLite → QueuePool（2.0+ 的默认值）；:memory: → SingletonThreadPool；共享内存数据库／测试 → StaticPool。服务端数据库使用 QueuePool。不要假设各处使用的连接池都相同。
• SQLite 会静默忽略 with_for_update。 它不会报错——只是不会加锁。在 SQLite 上“正常运行”的代码可能出现竞态；请在实际目标引擎上验证。
• 异步提交后延迟加载会引发异常。 异步提交后访问尚未加载的关系／属性会触发隐式 I/O，而异步操作不允许这样做。请使用 expire_on_commit=False，和／或预先加载。
• .execute(select(Model)) 返回的是行，而不是实体。 忘记调用 .scalars() 会得到单元素元组（(Job,)），而不是 Job。
• MySQL 的 SKIP LOCKED 需要 8.0+。 在 5.7 上它会导致语法错误，而不是被静默忽略。

反模式

• 不要把 create_all() 当作迁移策略发布，理由是“这样更简单”——它无法演进架构。请使用 alembic。
• 不要编写假装一个 API 能适配所有方言的“通用 upsert”辅助函数——excluded／ON CONFLICT（PG／SQLite）和 inserted／ON DUPLICATE KEY UPDATE（MySQL）确实不同。请按方言分支处理。
• 不要只依赖 with_for_update 来构建领取下一个任务的循环，并假设 SQLite 会锁定行——它不会。这是 sql-job-queue 的工作。
• 不要因为教程或源技能展示了旧式 Column =／declarative_base() 写法，就采用它——这是 2.0 之前的写法，而且会失去类型信息。
• 不要为整个应用打开一个长期存活的 Session（“为了避免开销”）——会话开销很低；长期存活的会话会泄漏过时的身份映射状态，并破坏并发。
• 不要为每个请求／任务创建一个引擎——这样会失去连接池的优势，并耗尽连接。

输出

一个可用的多方言 SQLAlchemy 2.x 数据层（或所需的部分）：每个进程一个配置了正确 URL 和连接池的 Engine／AsyncEngine、带类型的 DeclarativeBase／Mapped 模型、简短的会话级事务、给出选择理由的同步或异步驱动，以及针对各项跨方言易踩坑问题的方言分支。下游层会使用此产物——用于演进架构的迁移工具（alembic），或组合这些基础构件的更高层模式（例如 sql-job-queue）。

相关内容

• alembic — 架构迁移／受管理的 ALTER 变更（包括 SQLite 没有 ALTER 时批量移动并复制的易踩坑问题）。此技能仅在开发／测试引导阶段使用 create_all()，并将受管理的变更交由它处理。
• sql-job-queue — 任务存储／调度器模式：就绪集查询、原子租约循环、心跳／崩溃恢复、公平份额，以及在 with_for_update 不会加锁的 SQLite 上该如何处理。此技能仅介绍 with_for_update(skip_locked=...) 基础构件及其支持矩阵。
• pydantic-v2 — 应用边界处的数据验证／序列化（而非 ORM 列类型）。

渐进式披露

• references/sqlalchemy-extras.md — 跨方言易踩坑问题的更完整示例（端到端的方言专属 upsert、JSON／JSONB、隔离级别、StaticPool 测试设置、用于 RETURNING 的 Core 降级方案）。**在以下情况下加载：**实现具体易踩坑项，且正文中的代码片段不够用时。
• references/sources.md — 此技能的研究溯源信息。**在以下情况下加载：**验证某项声明的来源，或进行新一轮审查时。

正文预算

• description ≤ 1,024 个字符。
• 正文 ≤ 约 500 行 / 5,000 个 token。
• 每个参考文件：超过 10k 个 token 时发出警告，超过 25k 个 token 时报错。参考文件总计：超过 25k 个 token 时发出警告，超过 50k 个 token 时报错。