SuperAPI AI 工具集

sqlalchemy

编程与开发

查看项目

更新于 10-15 20:07

技能介绍

用途:指导智能体用 SQLAlchemy 2.x(2.0/2.1)搭建可移植的关系数据层,让模型、会话和查询在 SQLite、PostgreSQL、MySQL 上无需改动即可运行。以类型化 2.x ORM(DeclarativeBase、Mapped[...]、mapped_column(...))为主,仅在 ORM 无法可移植表达时(方言专属 upsert、RETURNING)下沉到 Core。同步优先,异步作为简短补充。

主要能力:选择方言与驱动并编写连接 URL(如 sqlite+pysqlite:///./app.db、postgresql+psycopg://、mysql+pymysql://);每进程构建一个 Engine 并配置连接池(pool_pre_ping、pool_size、max_overflow);定义类型化模型;用短生命周期会话与显式事务执行读写;同步/异步驱动选择矩阵;跨方言差异处理,包括行锁 with_for_update(skip_locked=...)、通用 JSON、按方言分支的 upsert(PG/SQLite 用 on_conflict_do_update,MySQL 用 on_duplicate_key_update)、整数主键自增、事务隔离级别设置。

使用方式:按工作流执行——选方言与驱动写 URL;每进程建一个 Engine;定义 DeclarativeBase 与 Mapped 模型;开发/测试用 Base.metadata.create_all(engine) 引导建表;每个工作单元用 with Session(engine) as session, session.begin() 开启事务;读取用 session.get(Model, pk) 或 session.execute(select(...)).scalars().all()。异步用 create_async_engine 与 async_sessionmaker,并设 expire_on_commit=False。

适用场景:面向 SQLite、PostgreSQL、MySQL 中一个或多个方言的应用数据层;选择 DBAPI 驱动;配置连接池、事务范围或隔离级别;排查“在 SQLite 可用、在 Postgres 失败”的差异问题。

限制:不用于 schema 迁移(交给 alembic);不用于任务队列/调度器(交给 sql-job-queue);不用于 Pydantic 校验;不用于其他 ORM 或原生 DBAPI。硬性规则:只用 2.x 类型化 ORM,每进程一个 Engine,会话短生命周期且不跨线程/异步任务共享,create_all() 仅限开发测试引导,必须按方言分支处理差异。注意 SQLite 无行级锁,with_for_update 静默失效;MySQL SKIP LOCKED 需 8.0+;异步提交后惰性加载会报错。