# pbatis **Repository Path**: pbatis/pbatis ## Basic Information - **Project Name**: pbatis - **Description**: pbatis orm framework - **Primary Language**: Unknown - **License**: MIT - **Default Branch**: main - **Homepage**: None - **GVP Project**: No ## Statistics - **Stars**: 0 - **Forks**: 0 - **Created**: 2026-10-12 - **Last Updated**: 2026-10-13 ## Categories & Tags **Categories**: Uncategorized **Tags**: None ## README PyBatis ======= .. image:: docs/pbatis-logo.png :alt: PyBatis ORM Logo :align: center :width: 280px Python 数据库工具与对象关系映射（ORM）框架。 提供 SQL 表达式语言、连接池、Schema 元数据、声明式 ORM 和跨方言的数据库访问能力，支持同步与异步两种编程模型，覆盖 SQLite、MySQL/MariaDB、PostgreSQL、Oracle、Microsoft SQL Server 等主流关系型数据库。 ---- 安装 ---- .. code-block:: bash pip install pbatis 按需安装特定方言驱动： .. code-block:: bash # SQLite（标准库自带 pysqlite，无需额外安装） pip install pbatis[aiosqlite] # 异步 SQLite pip install pbatis[sqlcipher] # 加密 SQLite # MySQL / MariaDB pip install pbatis[mysql] # mysqlclient pip install pbatis[pymysql] # PyMySQL（纯 Python） pip install pbatis[mysql-connector] # mysql-connector-python pip install pbatis[mariadb-connector]# mariadb 官方驱动 pip install pbatis[aiomysql] pip install pbatis[asyncmy] # PostgreSQL pip install pbatis[postgresql] # psycopg 3 pip install pbatis[postgresql-psycopg2binary] # psycopg2-binary pip install pbatis[postgresql-pg8000] pip install pbatis[postgresql-asyncpg] # Oracle pip install pbatis[oracle] # oracledb（推荐） pip install pbatis[oracle-cxoracle] # cx_Oracle # Microsoft SQL Server pip install pbatis[mssql] # pyodbc pip install pbatis[mssql-pymssql] pip install pbatis[mssql-python] # mssql-python # 异步支持（greenlet） pip install pbatis[asyncio] 要求 Python 3.11 及以上。 ---- 快速开始 -------- .. code-block:: python from pbatis import ( create_engine, MetaData, Table, Column, Integer, String, ForeignKey, select, insert, update, delete, ) # 1. 建表（Schema） metadata = MetaData() users = Table( "users", metadata, Column("id", Integer, primary_key=True), Column("name", String(50), nullable=False), Column("email", String(120), unique=True), ) addresses = Table( "addresses", metadata, Column("id", Integer, primary_key=True), Column("user_id", Integer, ForeignKey("users.id")), Column("city", String(80)), ) # 2. 创建引擎（Engine = 连接池 + 方言） engine = create_engine("sqlite:///:memory:", echo=True) # 3. 建表 metadata.create_all(engine) # 4. 插入数据 with engine.begin() as conn: conn.execute(insert(users).values(name="张三", email="a@example.com")) conn.execute(insert(addresses).values(user_id=1, city="深圳")) # 5. 查询 with engine.connect() as conn: for row in conn.execute(select(users)): print(row.id, row.name, row.email) ---- 核心概念 -------- Engine —— 数据库引擎 ~~~~~~~~~~~~~~~~~~~~~ ``Engine`` 是 PyBatis 访问数据库的入口，内部维护一个连接池。 .. code-block:: python from pbatis import create_engine # SQLite 内存库（测试用） engine = create_engine("sqlite:///:memory:") # SQLite 文件库 engine = create_engine("sqlite:///data/app.db") # MySQL engine = create_engine("mysql+pymysql://user:password@localhost:3306/mydb?charset=utf8mb4") # PostgreSQL engine = create_engine("postgresql+psycopg://user:password@localhost:5432/mydb") # 通用参数 engine = create_engine( "postgresql+psycopg://localhost/mydb", echo=True, # 打印 SQL pool_size=5, # 连接池大小 max_overflow=10, # 超出池大小时允许的临时连接 pool_recycle=3600, # 连接回收秒数 pool_pre_ping=True, # 取出连接前 ping 一下 ) Connection —— 数据库连接 ~~~~~~~~~~~~~~~~~~~~~~~~~ ``Connection`` 由 Engine 提供，支持上下文管理器自动释放。 .. code-block:: python # 自动提交（事务） with engine.begin() as conn: conn.execute(insert(users).values(name="李四")) conn.execute(insert(addresses).values(user_id=2, city="广州")) # 退出 with 块自动 commit；异常自动 rollback # 手动事务 with engine.connect() as conn: conn.execute(insert(users).values(name="王五")) conn.commit() # 显式提交 # 或 conn.rollback() # 同一连接上多次执行 with engine.connect() as conn: result = conn.execute(select(users)) for row in result: print(row) # 异步连接 from pbatis.ext.asyncio import create_async_engine async with create_async_engine("sqlite+aiosqlite:///:memory:").begin() as conn: await conn.execute(insert(users).values(name="赵六")) Schema —— 表与列定义 ~~~~~~~~~~~~~~~~~~~~ 使用 ``Table`` / ``Column`` 直接定义；或在 ORM 模式下用 ``declarative_base()`` 声明。 .. code-block:: python from pbatis import ( MetaData, Table, Column, Integer, String, ForeignKey, DateTime, Numeric, Boolean, func, ) metadata = MetaData() # 基础类型 products = Table( "products", metadata, Column("id", Integer, primary_key=True, autoincrement=True), Column("sku", String(32), unique=True, nullable=False), Column("name", String(200), nullable=False), Column("price", Numeric(10, 2), default=0), Column("active", Boolean, default=True), Column("created_at", DateTime, server_default=func.now()), Column("updated_at", DateTime, onupdate=func.now()), ) # 主键约束、唯一约束、检查约束 from pbatis import UniqueConstraint, CheckConstraint, Index accounts = Table( "accounts", metadata, Column("id", Integer, primary_key=True), Column("email", String(120)), Column("age", Integer), UniqueConstraint("email", name="uq_accounts_email"), CheckConstraint("age >= 0", name="ck_accounts_age"), Index("ix_accounts_email", "email"), ) # 复合主键 order_items = Table( "order_items", metadata, Column("order_id", Integer, primary_key=True), Column("line_no", Integer, primary_key=True), Column("sku", String(32)), ) # 跨表外键 orders = Table( "orders", metadata, Column("id", Integer, primary_key=True), Column("user_id", Integer, ForeignKey("users.id", ondelete="CASCADE")), Column("total", Numeric(10, 2)), ) 数据类型速查 ~~~~~~~~~~~~ .. list-table:: :header-rows: 1 :widths: 20 30 50 * - 类别 - 类型 - 适用场景 * - 整数 - ``Integer``、``BigInteger``、``SmallInteger`` - 主键、外键、计数 * - 字符串 - ``String``、``Text``、``Unicode``、``UnicodeText`` - 短/长文本，Unicode 显式指定字符集 * - 浮点 - ``Float``、``Numeric``、``Double`` - 金额用 ``Numeric`` 避免二进制浮点误差 * - 日期时间 - ``Date``、``Time``、``DateTime``、``Interval`` - 时间戳、时段 * - 布尔 - ``Boolean`` - 真假标志位 * - JSON - ``JSON`` - 文档型字段（方言支持差异见下） * - 二进制 - ``LargeBinary`` - 字节流、BLOB * - 枚举 - ``Enum``、``PickleType`` - 有限集合 * - 数组（PG） - ``ARRAY(Integer)`` - PostgreSQL 原生数组 * - 自定义 - ``TypeDecorator``、``UserDefinedType`` - 包装或自定义类型映射 ---- SQL 表达式语言（Core） ---------------------- 增（INSERT） ~~~~~~~~~~~~ .. code-block:: python from pbatis import insert # 单行插入 stmt = insert(users).values(name="张三", email="a@example.com") with engine.begin() as conn: result = conn.execute(stmt) print(result.inserted_primary_key) # 自动生成的主键 # 多行批量插入（性能更高） with engine.begin() as conn: conn.execute(insert(users), [ {"name": "李四", "email": "b@example.com"}, {"name": "王五", "email": "c@example.com"}, {"name": "赵六", "email": "d@example.com"}, ]) # INSERT ... SELECT（从另一张表查出来再插入） with engine.begin() as conn: conn.execute( insert(archived_users).select().where(users.c.active == False) ) # INSERT ... ON CONFLICT（PG）/ ON DUPLICATE KEY（MySQL）/ MERGE（Oracle） from pbatis.dialects.postgresql import insert as pg_insert stmt = pg_insert(users).values(id=1, name="冲突", email="x@x.com") upsert = stmt.on_conflict_do_update( index_elements=[users.c.id], set_={"name": stmt.excluded.name}, ) with engine.begin() as conn: conn.execute(upsert) # MySQL 等价写法 from pbatis.dialects.mysql import insert as mysql_insert stmt = mysql_insert(users).values(id=1, name="冲突", email="x@x.com") upsert = stmt.on_duplicate_key_update(name=stmt.inserted.name) # 返回值（RETURNING） from pbatis import insert stmt = insert(users).values(name="回写").returning(users.c.id, users.c.name) with engine.connect() as conn: for row in conn.execute(stmt): print(row) 删（DELETE） ~~~~~~~~~~~~ .. code-block:: python from pbatis import delete # 按条件删除 with engine.begin() as conn: conn.execute(delete(users).where(users.c.id == 5)) # 多条件 with engine.begin() as conn: conn.execute( delete(users) .where(users.c.active == False) .where(users.c.created_at < "2020-01-01") ) # RETURNING（PostgreSQL / SQLite / Oracle 21c+） stmt = delete(users).where(users.c.id == 5).returning(users.c.id) with engine.connect() as conn: for row in conn.execute(stmt): print("deleted", row.id) 改（UPDATE） ~~~~~~~~~~~~ .. code-block:: python from pbatis import update # 单字段 with engine.begin() as conn: conn.execute( update(users) .where(users.c.id == 1) .values(name="张三丰") ) # 多字段 with engine.begin() as conn: conn.execute( update(users) .where(users.c.email == "old@example.com") .values(name="新名", active=False, updated_at=func.now()) ) # 批量按表达式更新 from pbatis import case with engine.begin() as conn: conn.execute( update(products) .where(products.c.category == "books") .values(price=case( (products.c.price < 50, products.c.price * 1.1), (products.c.price < 100, products.c.price * 1.05), else_=products.c.price, )) ) # RETURNING stmt = ( update(users) .where(users.c.id == 1) .values(name="新") .returning(users.c.id, users.c.name) ) with engine.connect() as conn: for row in conn.execute(stmt): print(row) 查（SELECT） ~~~~~~~~~~~~ .. code-block:: python from pbatis import select, and_, or_, not_ # 基础查询 stmt = select(users.c.id, users.c.name).where(users.c.active == True) with engine.connect() as conn: for row in conn.execute(stmt): print(row.id, row.name) # 排序 stmt = select(users).order_by(users.c.created_at.desc(), users.c.id.asc()) # LIMIT / OFFSET stmt = select(users).limit(10).offset(20) # DISTINCT stmt = select(users.c.city).distinct() # 聚合 from pbatis import func stmt = select( users.c.city, func.count(users.c.id).label("cnt"), func.avg(users.c.age).label("avg_age"), ).group_by(users.c.city).having(func.count(users.c.id) > 5) # 子查询 subq = select(func.max(orders.c.total)).scalar_subquery() stmt = select(orders).where(orders.c.total == subq) # UNION from pbatis import union, union_all stmt = union( select(users.c.email).where(users.c.active == True), select(addresses.c.email).where(addresses.c.primary == True), ) # CTE（公共表表达式） cte = ( select(orders.c.user_id, func.sum(orders.c.total).label("total")) .group_by(orders.c.user_id) .cte("user_totals") ) stmt = select(users.c.name, cte.c.total).join(cte, cte.c.user_id == users.c.id) ---- 查询条件与过滤 -------------- WHERE 子句里能写什么——一次性讲清。 比较运算符 ~~~~~~~~~~ .. code-block:: python from pbatis import select # 等于 = select(users).where(users.c.name == "张三") # 渲染：WHERE name = '张三' # 不等于（两种写法等价） select(users).where(users.c.name != "张三") select(users).where(users.c.name.op("!=")("张三")) # 大于 / 小于 / 大于等于 / 小于等于 select(products).where(products.c.price > 100) # 大于 select(products).where(products.c.price < 100) # 小于 select(products).where(products.c.price >= 100) # 大于等于 select(products).where(products.c.price <= 100) # 小于等于 # 区间 BETWEEN（闭区间，含两端） from pbatis import between select(products).where(products.c.price.between(50, 200)) # 渲染：WHERE price BETWEEN 50 AND 200 # 排除区间（NOT BETWEEN） select(products).where(~products.c.price.between(50, 200)) # 同值多个字段（row-value 比较） select(orders).where( tuple_(orders.c.region, orders.c.priority) == tuple_("CN", 1) ) 运算符速查 ^^^^^^^^^^ .. list-table:: :header-rows: 1 :widths: 20 30 50 * - 运算符 - Python 写法 - 生成的 SQL * - 等于 - ``col == "x"`` - ``col = 'x'`` * - 不等于 - ``col != "x"`` - ``col <> 'x'`` * - 大于 / 小于 - ``col > 10``、``col < 10`` - ``col > 10``、``col < 10`` * - 大于等于 / 小于等于 - ``col >= 10``、``col <= 10`` - ``col >= 10``、``col <= 10`` * - BETWEEN - ``col.between(a, b)`` - ``col BETWEEN a AND b`` * - IN - ``col.in_([...])`` - ``col IN (...)`` * - NOT IN - ``~col.in_([...])`` - ``col NOT IN (...)`` * - IS NULL - ``col.is_(None)`` - ``col IS NULL`` * - IS NOT NULL - ``col.isnot(None)`` - ``col IS NOT NULL`` * - LIKE - ``col.like("%foo%")`` - ``col LIKE '%foo%'`` * - ILIKE（PG） - ``col.ilike("%foo%")`` - ``col ILIKE '%foo%'`` * - 真值 - ``col.is_(True)`` - ``col IS true`` 逻辑组合 AND / OR / NOT ~~~~~~~~~~~~~~~~~~~~~~~~ .. code-block:: python from pbatis import and_, or_, not_ # 1. 隐式 AND（链式 .where()） select(users).where( users.c.age > 18, users.c.city == "深圳", users.c.active == True, ) # 等价于：WHERE age > 18 AND city = '深圳' AND active = true # 2. 显式 and_(...) select(users).where(and_( users.c.age > 18, users.c.city == "深圳", )) # 3. OR select(users).where(or_( users.c.city == "深圳", users.c.city == "广州", )) # 渲染：WHERE city = '深圳' OR city = '广州' # 4. 混合 select(users).where(and_( users.c.age > 18, or_( users.c.city == "深圳", users.c.city == "广州", ), users.c.active == True, )) # 渲染：WHERE age > 18 AND (city = '深圳' OR city = '广州') AND active = true # 5. NOT（取反，前缀 ~） select(users).where(~users.c.active) # 活跃 = false select(users).where(not_(users.c.age > 18)) # 18 岁以下 # 6. 用布尔对象直接取反 from pbatis import true, false, null select(users).where(users.c.active == false) select(users).where(users.c.active.is_not(True)) # 7. any_/all_ 配合子查询（PG、MySQL 8+） from pbatis import any_, all_ select(products).where(products.c.price < any_( select(orders.c.total).where(orders.c.user_id == 1) )) NULL 处理 ~~~~~~~~~ .. code-block:: python # IS NULL select(users).where(users.c.email.is_(None)) # 渲染：WHERE email IS NULL # IS NOT NULL select(users).where(users.c.email.isnot(None)) # 渲染：WHERE email IS NOT NULL # 不要用 == None（不会生成 IS NULL） # select(users).where(users.c.email == None) # 错：会变成 email = NULL（永远不匹配） # COALESCE（取第一个非 NULL） from pbatis import func select(func.coalesce(users.c.nickname, users.c.name, "匿名")) # NULLIF（两值相等返回 NULL） select(func.nullif(users.c.status, "deleted")) # 字段是否为 NULL 的布尔返回（PG / MySQL） select(users).where(users.c.deleted_at.is_(None)) # 未删除 select(users).where(users.c.deleted_at.isnot(None)) # 已删除 字符串匹配 LIKE / ILIKE / REGEX ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ .. code-block:: python # 模糊匹配：% 任意字符、_ 单字符 select(users).where(users.c.name.like("张%")) # 张某、张三丰 select(users).where(users.c.name.like("%三%")) # 包含"三" select(users).where(users.c.name.like("张_")) # 张三（两字） # 不区分大小写（PG 用 ilike，MySQL/SQLite 用 lower()） select(users).where(users.c.name.ilike("%ZHANG%")) # PG select(users).where(users.c.name.lower().like("%zhang%")) # 跨方言 # 取反 select(users).where(~users.c.name.like("张%")) # NOT LIKE # 转义 LIKE 通配符 select(users).where( users.c.filename.like("%10\%%", escape="\\") ) # 正则（PG、Oracle、MySQL 8+） select(users).where(users.c.email.op("~")("^[a-z]+@")) # PG select(users).where(users.c.name.op("REGEXP")("^张")) # MySQL # 全文搜索（PG） from pbatis.dialects.postgresql import TSVECTOR stmt = ( select(articles) .where(articles.c.searchable.op("@@")(func.to_tsquery("python & sql"))) ) # 全文搜索（MySQL） stmt = select(articles).where(articles.c.body.op("MATCH")("python sql")) 集合操作 IN / NOT IN / ANY / ALL ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ .. code-block:: python # IN：值在列表内 select(users).where(users.c.city.in_(["深圳", "广州", "上海"])) # 渲染：WHERE city IN ('深圳', '广州', '上海') # NOT IN select(users).where(~users.c.city.in_(["北京", "上海"])) # IN 子查询 select(users).where(users.c.id.in_( select(orders.c.user_id).where(orders.c.total > 1000) )) # 元组 IN（PG、MySQL） select(orders).where( tuple_(orders.c.region, orders.c.status).in_([ ("CN", "paid"), ("HK", "shipped"), ]) ) # 任意一个满足（= ANY IN） select(products).where(products.c.tag_id == any_([1, 2, 3])) # 渲染：WHERE tag_id = ANY(ARRAY[1, 2, 3]) # 全部满足（ALL） select(products).where(products.c.price > all_([100, 200, 300])) 排序 ORDER BY ~~~~~~~~~~~~~ .. code-block:: python from pbatis import asc, desc, nulls_first, nulls_last # 单列升序 / 降序 select(users).order_by(users.c.name.asc()) select(users).order_by(users.c.name.desc()) # 简写 select(users).order_by(users.c.name) # 默认 ASC select(users).order_by(users.c.created_at.desc()) # 多列排序（按列表顺序生效） select(users).order_by( users.c.active.desc(), # 活跃用户排前 users.c.created_at.desc(), # 同状态按时间倒序 users.c.id.asc(), # 同时间按 id 升序 ) # NULL 位置（PG、Oracle、MSSQL；MySQL 8+） select(users).order_by(users.c.last_login.desc().nulls_last()) # last_login IS NULL 排末尾 select(users).order_by(users.c.last_login.desc().nulls_first()) # last_login IS NULL 排前面 # 表达式排序 select(products).order_by( func.length(products.c.name).asc(), # 按名称长度排 ) select(users).order_by( func.random(), # 随机（SQLite/PG） ) select(users).order_by( func.rand(), # 随机（MySQL） ) # CASE 排序（自定义优先级） from pbatis import case priority = case( (users.c.role == "admin", 1), (users.c.role == "vip", 2), else_=3, ) select(users).order_by(priority.asc(), users.c.id.asc()) 分页 LIMIT / OFFSET ~~~~~~~~~~~~~~~~~~~~ .. code-block:: python # 基础分页 stmt = select(users).order_by(users.c.id).limit(10).offset(20) # 跳过前 20 条，取 10 条 # 仅取前 N 条 select(users).order_by(users.c.id).limit(5) # 切片（等价 limit + offset） select(users).order_by(users.c.id).slice(20, 30) # skip 20, take 10 # 大数据集分页——用 WHERE 替代 OFFSET（性能更好） # 慢：OFFSET 越大越慢 # select(users).where(...).offset(1000000) # 快：键集分页（seek pagination） last_id = 1000000 select(users).where(users.c.id > last_id).order_by(users.c.id).limit(10) # 流式读取（处理超大数据集） with engine.connect() as conn: result = conn.execution_options(stream_results=True).execute( select(events).order_by(events.c.id) ) for row in result: process(row) DISTINCT 去重 ~~~~~~~~~~~~~ .. code-block:: python # 单列去重 select(users.c.city).distinct() # 渲染：SELECT DISTINCT city # 多列联合去重 select(users.c.city, users.c.gender).distinct() # city + gender 都相同才算重复 # 配合聚合 select(func.count(func.distinct(users.c.city))).select_from(users) # 不同城市数量 聚合与 HAVING ~~~~~~~~~~~~~~ .. code-block:: python from pbatis import func # 基础聚合 select(func.count(users.c.id)) # 总行数 select(func.count(func.distinct(users.c.city))) # 不同城市数 select(func.sum(orders.c.total)) # 总额 select(func.avg(orders.c.total)) # 平均 select(func.min(orders.c.total), func.max(orders.c.total)) # 最小/最大 # 分组（GROUP BY） stmt = ( select( users.c.city, func.count(users.c.id).label("cnt"), ) .group_by(users.c.city) ) # SELECT city, COUNT(id) AS cnt FROM users GROUP BY city # 多列分组 stmt = ( select( users.c.city, users.c.gender, func.count(users.c.id), ) .group_by(users.c.city, users.c.gender) ) # HAVING（分组后过滤） stmt = ( select(users.c.city, func.count(users.c.id).label("cnt")) .group_by(users.c.city) .having(func.count(users.c.id) > 100) ) # HAVING COUNT(id) > 100 # WHERE + GROUP + HAVING stmt = ( select(users.c.city, func.avg(users.c.age).label("avg_age")) .where(users.c.active == True) .group_by(users.c.city) .having(func.avg(users.c.age) > 30) .order_by(func.avg(users.c.age).desc()) ) EXISTS 与子查询 ~~~~~~~~~~~~~~~ .. code-block:: python from pbatis import select, exists, not_ # EXISTS：相关子查询有结果即为真 stmt = select(users).where( exists().where(orders.c.user_id == users.c.id) ) # 找出有订单的用户 # NOT EXISTS stmt = select(users).where( ~exists().where(orders.c.user_id == users.c.id) ) # 找出没有任何订单的用户 # 标量子查询（返回单值） subq = select(func.max(orders.c.total)).scalar_subquery() stmt = select(orders).where(orders.c.total == subq) # 找出金额等于最大金额的订单 # 子查询作为列 order_count_subq = ( select(func.count(orders.c.id)) .where(orders.c.user_id == users.c.id) .scalar_subquery() ) stmt = select(users.c.name, order_count_subq.label("order_cnt")) # SELECT name, (SELECT COUNT(*) FROM orders WHERE user_id = users.id) AS order_cnt 时间范围查询 ~~~~~~~~~~~~ .. code-block:: python from datetime import date, datetime, timedelta from pbatis import func, between # 具体日期 stmt = select(orders).where(orders.c.created_at >= datetime(2024, 1, 1)) # 当天 today = date.today() stmt = select(orders).where(func.date(orders.c.created_at) == today) # 本周 stmt = select(orders).where( orders.c.created_at >= func.date_sub(func.now(), func.interval(7, "day")) ) # MySQL stmt = select(orders).where( orders.c.created_at >= func.now() - timedelta(days=7) ) # 跨方言 # BETWEEN 时间范围 start = datetime(2024, 1, 1) end = datetime(2024, 1, 31, 23, 59, 59) stmt = select(orders).where(orders.c.created_at.between(start, end)) # 日期函数 stmt = select(orders).where(func.year(orders.c.created_at) == 2024) stmt = select(orders).where(func.month(orders.c.created_at) == 6) stmt = select(orders).where(func.date(orders.c.created_at) == "2024-06-01") # PG：date_trunc 按月分桶 stmt = ( select( func.date_trunc("month", orders.c.created_at).label("month"), func.sum(orders.c.total), ) .group_by("month") ) JSON 字段查询 ~~~~~~~~~~~~~~ .. code-block:: python # PG JSONB from pbatis.dialects.postgresql import JSONB users = Table("users", m, Column("profile", JSONB)) # 顶层 key stmt = select(users).where(users.c.profile["city"].astext == "深圳") # 嵌套路径 stmt = select(users).where( users.c.profile["address"]["zip"].astext == "518000" ) # 包含（PG JSONB） stmt = select(users).where(users.c.profile.contains({"vip": True})) # 包含任一 key stmt = select(users).where( users.c.profile.has_key("preferences") ) # 数组长度（PG JSONB） stmt = select(users).where( func.jsonb_array_length(users.c.profile["tags"]) > 3 ) # MySQL JSON stmt = select(users).where( users.c.profile["$.city"].astext == "深圳" ) # 注意：MySQL 路径以 $. 开头 # SQLite JSON stmt = select(users).where( func.json_extract(users.c.profile, "$.city") == "深圳" ) 查询案例（业务场景 → SQL） ~~~~~~~~~~~~~~~~~~~~~~~~~~ 30 个"问业务问题"→"对应 PyBatis 写法"。 .. list-table:: :header-rows: 1 :widths: 35 65 * - 业务问句 - 写法 * - 找出所有 30 岁以上的深圳用户 - ``select(users).where(users.c.age > 30, users.c.city == "深圳")`` * - 名字里包含"张"字的用户 - ``select(users).where(users.c.name.like("%张%"))`` * - 没有邮箱的用户 - ``select(users).where(users.c.email.is_(None))`` * - 邮箱不为空的活跃用户 - ``select(users).where(users.c.email.isnot(None), users.c.active == True)`` * - 30-50 岁之间的用户 - ``select(users).where(users.c.age.between(30, 50))`` * - 订单金额 100-500 之间且已支付 - ``select(orders).where(orders.c.total.between(100, 500), orders.c.status == "paid")`` * - 查询在深圳或广州的用户 - ``select(users).where(users.c.city.in_(["深圳", "广州"]))`` * - 排除黑名单用户 - ``select(users).where(~users.c.id.in_(blacklist_ids))`` * - 有订单记录的用户 - ``select(users).where(exists().where(orders.c.user_id == users.c.id))`` * - 从未下单的用户 - ``select(users).where(~exists().where(orders.c.user_id == users.c.id))`` * - 订单金额最高的 10 单 - ``select(orders).order_by(orders.c.total.desc()).limit(10)`` * - 第 21-40 条用户（按 id 升序） - ``select(users).order_by(users.c.id).offset(20).limit(20)`` * - 一个月内注册的新用户 - ``select(users).where(users.c.created_at >= func.now() - timedelta(days=30))`` * - 不同城市数量 - ``select(func.count(func.distinct(users.c.city)))`` * - 每个城市的用户数 - ``select(users.c.city, func.count(users.c.id)).group_by(users.c.city)`` * - 用户数大于 1000 的城市 - ``.group_by(users.c.city).having(func.count(users.c.id) > 1000)`` * - 价格最高的前 3 件商品 - ``select(products).order_by(products.c.price.desc()).limit(3)`` * - 按价格区间分组（0-100、100-500、500+） - 用 ``case()`` 配合 ``func.sum()`` 实现 * - 名字长度大于 5 的用户 - ``select(users).where(func.length(users.c.name) > 5)`` * - 邮箱域名是 @example.com - ``select(users).where(users.c.email.like("%@example.com"))`` * - 库存为零的商品 - ``select(products).where(products.c.stock == 0)`` * - 库存不为零且上架 - ``select(products).where(products.c.stock > 0, products.c.active == True)`` * - 模糊搜索（不区分大小写） - ``select(users).where(users.c.name.ilike("%query%"))`` (PG) 或 ``.lower().like()`` * - 找出金额大于平均值的订单 - ``select(orders).where(orders.c.total > select(func.avg(orders.c.total)).scalar_subquery())`` * - 每年订单总额 - ``select(func.extract("year", orders.c.created_at), func.sum(orders.c.total)).group_by(1)`` * - 找出 7 天内未登录的用户 - ``select(users).where(or_(users.c.last_login.is_(None), users.c.last_login < func.now() - timedelta(days=7)))`` * - 同一手机号只返回最新一条 - 用窗口函数 ``func.row_number().over(partition_by=..., order_by=...)`` * - 查询 JSON 字段 city = 深圳 - 见上文"JSON 字段查询" * - 关键字全文搜索 - 见上文"字符串匹配"末尾 * - 分组后取每组前 N 条 - 用 CTE + 窗口函数 ``row_number()`` 分组内 Top-N（窗口函数） ~~~~~~~~~~~~~~~~~~~~~~~~~ .. code-block:: python from pbatis import over, func # ROW_NUMBER：每组从 1 开始编号 rn = func.row_number().over( partition_by=orders.c.user_id, # 分组字段 order_by=orders.c.total.desc(), # 组内排序 ).label("rn") # 找出每个用户金额最高的 1 单 subq = select(orders.c.id, orders.c.user_id, orders.c.total, rn).subquery() stmt = select(subq).where(subq.c.rn == 1) # RANK：并列同名次会跳号 rk = func.rank().over( partition_by=orders.c.user_id, order_by=orders.c.total.desc(), ) stmt = select(orders.c.user_id, orders.c.total, rk.label("rk")).order_by("rk") # DENSE_RANK：并列不跳号 drk = func.dense_rank().over( partition_by=orders.c.user_id, order_by=orders.c.total.desc(), ) # 移动平均 ma = func.avg(orders.c.total).over( partition_by=orders.c.user_id, order_by=orders.c.created_at, rows=(None, 6), # 7 天窗口（含当前行） ) # 累计求和 cumsum = func.sum(orders.c.total).over( partition_by=orders.c.user_id, order_by=orders.c.created_at, ) # 配合 WITH 子句取每组前 3 cte = ( select( orders.c.id, orders.c.user_id, orders.c.total, func.row_number().over( partition_by=orders.c.user_id, order_by=orders.c.total.desc(), ).label("rn"), ) .cte("ranked") ) stmt = select(cte).where(cte.c.rn <= 3).order_by(cte.c.user_id, cte.c.rn) FOR UPDATE / 锁 ~~~~~~~~~~~~~~~ .. code-block:: python from pbatis import select # 行级排他锁（PG、MySQL InnoDB、Oracle） stmt = select(accounts).where(accounts.c.id == 1).with_for_update() # 渲染：SELECT ... FOR UPDATE # 跳过已被锁的行 stmt = select(accounts).where(accounts.c.id == 1).with_for_update(skip_locked=True) # 渲染：SELECT ... FOR UPDATE SKIP LOCKED # 共享锁 stmt = select(accounts).where(accounts.c.id == 1).with_for_update(read=True) # 渲染：SELECT ... FOR SHARE（PG）/ LOCK IN SHARE MODE（MySQL） # 在事务中使用 with engine.begin() as conn: acct = conn.execute(select(accounts).where(accounts.c.id == 1).with_for_update()).first() acct.balance -= 100 conn.execute(update(accounts).where(accounts.c.id == 1).values(balance=acct.balance)) 性能提示 ~~~~~~~~ .. code-block:: python # 1. 优先用 = 而不是 LIKE '%xx%'（前缀 % 用不到索引） select(users).where(users.c.name.like("张%")) # OK select(users).where(users.c.name.like("%张")) # 全表扫描 # 2. 优先用 IN 而不是 OR select(users).where(users.c.id.in_([1, 2, 3])) # OK select(users).where(or_(users.c.id == 1, users.c.id == 2)) # 不如 IN # 3. 大量分页用键集分页而不是 OFFSET # 慢：select(users).offset(1000000) # 快：select(users).where(users.c.id > last_seen_id).limit(10) # 4. 配合索引字段排序 # ORDER BY 用索引列（如主键），避免 filesort select(users).order_by(users.c.id) # 走主键 select(users).order_by(users.c.email) # 走索引 select(users).order_by(func.rand()) # 全表扫描 # 5. 不要 SELECT *，只取需要的列 select(users.c.id, users.c.name) # OK select(users) # 多取字段 # 6. 大结果集用流式 conn.execution_options(stream_results=True).execute(stmt) ---- 联表查询 -------- JOIN 基础 ~~~~~~~~~ .. code-block:: python from pbatis import select, and_ # INNER JOIN stmt = ( select(users.c.name, addresses.c.city) .select_from(users) .join(addresses, addresses.c.user_id == users.c.id) ) # LEFT OUTER JOIN stmt = ( select(users.c.name, addresses.c.city) .select_from(users) .join(addresses, addresses.c.user_id == users.c.id, isouter=True) ) # FULL OUTER JOIN（PG / Oracle） stmt = select(users, addresses).join(addresses, ..., full=True) # 多表链式 JOIN stmt = ( select(orders) .join(users, users.c.id == orders.c.user_id) .join(addresses, addresses.c.user_id == users.c.id) ) # ON 子句 + WHERE stmt = ( select(users) .join(addresses, addresses.c.user_id == users.c.id) .where(addresses.c.city == "深圳") ) # 自连接 employees = Table("employees", metadata, Column("id", Integer, primary_key=True), Column("name", String(50)), Column("manager_id", Integer, ForeignKey("employees.id")), ) managers = employees.alias("managers") stmt = ( select(employees.c.name.label("emp"), managers.c.name.label("mgr")) .join(managers, managers.c.id == employees.c.manager_id) ) # 索引/表名访问 select(users).where(users.c["my-column"] == 1) # 列名含连字符时 ORM 关系映射 ~~~~~~~~~~~~ PyBatis 提供声明式 ORM，使用 ``declarative_base()`` 配合 ``relationship`` 定义一对一、一对多、多对多关系。 .. code-block:: python from pbatis.orm import ( declarative_base, relationship, Mapped, mapped_column, ) from pbatis import Integer, String, ForeignKey Base = declarative_base() 一对一（One-to-One） ~~~~~~~~~~~~~~~~~~~~ .. code-block:: python class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(Integer, primary_key=True) name: Mapped[str] = mapped_column(String(50)) # 一对一：通过 uselist=False profile: Mapped["Profile"] = relationship( "Profile", back_populates="user", uselist=False, # 关键：返回单对象而不是列表 cascade="all, delete-orphan", ) class Profile(Base): __tablename__ = "profiles" id: Mapped[int] = mapped_column(Integer, primary_key=True) user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), unique=True) bio: Mapped[str] = mapped_column(String(500)) user: Mapped["User"] = relationship("User", back_populates="profile") # 增 with Session(engine) as s: u = User(name="张三", profile=Profile(bio="工程师")) s.add(u) s.commit() # 查 with Session(engine) as s: u = s.query(User).filter_by(name="张三").one() print(u.profile.bio) # 直接访问，无需 join # 删（cascade 自动清理） s.delete(u) s.commit() 一对多（One-to-Many） ~~~~~~~~~~~~~~~~~~~~~ .. code-block:: python class Author(Base): __tablename__ = "authors" id: Mapped[int] = mapped_column(Integer, primary_key=True) name: Mapped[str] = mapped_column(String(80)) # 一对多 books: Mapped[list["Book"]] = relationship( "Book", back_populates="author", cascade="all, delete-orphan", order_by="Book.id", ) class Book(Base): __tablename__ = "books" id: Mapped[int] = mapped_column(Integer, primary_key=True) title: Mapped[str] = mapped_column(String(200)) author_id: Mapped[int] = mapped_column(ForeignKey("authors.id")) author: Mapped["Author"] = relationship("Author", back_populates="books") # 增 with Session(engine) as s: a = Author(name="莫言", books=[ Book(title="丰乳肥臀"), Book(title="蛙"), ]) s.add(a) s.commit() # 查（懒加载） with Session(engine) as s: a = s.query(Author).filter_by(name="莫言").one() for b in a.books: print(b.title) # 预加载（避免 N+1） from pbatis.orm import selectinload, joinedload stmt = select(Author).options(selectinload(Author.books)) for a in s.execute(stmt).scalars(): for b in a.books: print(b.title) # 反向查 stmt = select(Book).join(Author).where(Author.name == "莫言") # 过滤 stmt = ( select(Author) .where(Author.books.any(Book.title.like("%丰%"))) ) 多对多（Many-to-Many） ~~~~~~~~~~~~~~~~~~~~~~ .. code-block:: python # 关联表（association table） student_course = Table( "student_course", Base.metadata, Column("student_id", Integer, ForeignKey("students.id"), primary_key=True), Column("course_id", Integer, ForeignKey("courses.id"), primary_key=True), Column("enrolled_at", DateTime, server_default=func.now()), ) class Student(Base): __tablename__ = "students" id: Mapped[int] = mapped_column(Integer, primary_key=True) name: Mapped[str] = mapped_column(String(80)) courses: Mapped[list["Course"]] = relationship( "Course", secondary=student_course, # 关键：指定关联表 back_populates="students", ) class Course(Base): __tablename__ = "courses" id: Mapped[int] = mapped_column(Integer, primary_key=True) title: Mapped[str] = mapped_column(String(200)) students: Mapped[list["Student"]] = relationship( "Student", secondary=student_course, back_populates="courses", ) # 增 with Session(engine) as s: math = Course(title="高等数学") eng = Course(title="英语") s.add(Student(name="小明", courses=[math, eng])) s.commit() # 增关联（追加） with Session(engine) as s: stu = s.query(Student).filter_by(name="小明").one() physics = s.query(Course).filter_by(title="物理").one() stu.courses.append(physics) s.commit() # 删关联 stu.courses.remove(math) s.commit() # 关联表上额外的字段（enrolled_at）—— 使用 Association Object class StudentCourse(Base): __tablename__ = "student_course" student_id: Mapped[int] = mapped_column(ForeignKey("students.id"), primary_key=True) course_id: Mapped[int] = mapped_column(ForeignKey("courses.id"), primary_key=True) enrolled_at: Mapped[datetime] = mapped_column(DateTime) grade: Mapped[Optional[float]] = mapped_column(Numeric(5, 2)) class StudentV2(Base): __tablename__ = "students" id: Mapped[int] = mapped_column(Integer, primary_key=True) name: Mapped[str] = mapped_column(String(80)) enrollments: Mapped[list["StudentCourse"]] = relationship( "StudentCourse", back_populates="student", cascade="all, delete-orphan" ) courses: Mapped[list["Course"]] = relationship( "Course", secondary="student_course", viewonly=True, ) class CourseV2(Base): __tablename__ = "courses" id: Mapped[int] = mapped_column(Integer, primary_key=True) title: Mapped[str] = mapped_column(String(200)) enrollments: Mapped[list["StudentCourse"]] = relationship( "StudentCourse", back_populates="course", cascade="all, delete-orphan" ) StudentCourse.student = relationship(StudentV2, back_populates="enrollments") StudentCourse.course = relationship(CourseV2, back_populates="enrollments") 自引用关系 ~~~~~~~~~~ .. code-block:: python class Employee(Base): __tablename__ = "employees" id: Mapped[int] = mapped_column(Integer, primary_key=True) name: Mapped[str] = mapped_column(String(80)) manager_id: Mapped[Optional[int]] = mapped_column(ForeignKey("employees.id")) # 一对多自引用：所有下属 subordinates: Mapped[list["Employee"]] = relationship( "Employee", remote_side="Employee.id", # 指向自己 back_populates="manager", ) manager: Mapped[Optional["Employee"]] = relationship( "Employee", remote_side="Employee.id", back_populates="subordinates", ) # 查所有上级链 with Session(engine) as s: e = s.query(Employee).filter_by(name="小李").one() for mgr in [e.manager, e.manager.manager, e.manager.manager.manager]: if mgr: print(mgr.name) 继承映射（Inheritance Mapping） ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ .. code-block:: python from pbatis.orm import declared_attr # 单表继承（joined / concrete / single-table 三种） class Person(Base): __tablename__ = "people" id: Mapped[int] = mapped_column(Integer, primary_key=True) name: Mapped[str] = mapped_column(String(80)) type: Mapped[str] = mapped_column(String(20)) # 鉴别列 __mapper_args__ = { "polymorphic_on": "type", "polymorphic_identity": "person", } class Engineer(Person): __tablename__ = "engineers" id: Mapped[int] = mapped_column(ForeignKey("people.id"), primary_key=True) language: Mapped[str] = mapped_column(String(40)) __mapper_args__ = { "polymorphic_identity": "engineer", } Eager Loading 策略 ~~~~~~~~~~~~~~~~~~ .. code-block:: python from pbatis.orm import ( selectinload, joinedload, subqueryload, contains_eager, ) # selectinload：推荐，1 条额外 IN 查询 stmt = select(User).options(selectinload(User.addresses)) # joinedload：1 条 LEFT JOIN（适合 many-to-one / one-to-one） stmt = select(Address).options(joinedload(Address.user)) # subqueryload：1 条子查询 stmt = select(User).options(subqueryload(User.orders)) # contains_eager：手写 JOIN 但仍按 relationship 填充 stmt = ( select(User) .join(Address) .where(Address.city == "深圳") .options(contains_eager(User.addresses)) ) Session 与生命周期 ~~~~~~~~~~~~~~~~~~ .. code-block:: python from pbatis.orm import Session, sessionmaker # 1. 直接用 Session 类（每次创建） with Session(engine) as s: s.add(user) s.commit() # 2. 用 sessionmaker 工厂（推荐，绑定配置） SessionLocal = sessionmaker(engine, expire_on_commit=False) with SessionLocal() as s: s.add(user) s.commit() # 3. scoped_session（线程/请求级单例） from pbatis.orm import scoped_session SessionScoped = scoped_session(SessionLocal) s = SessionScoped() # 同线程复用同一 Session s.add(user) s.commit() SessionScoped.remove() # 结束线程时清理 # 4. 异步 Session from pbatis.ext.asyncio import async_sessionmaker, AsyncSession AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False) async with AsyncSessionLocal() as s: s.add(user) await s.commit() 查询 API ~~~~~~~~ .. code-block:: python # Query 1.x 风格（保留兼容） users = s.query(User).filter(User.age > 18).order_by(User.id).all() one_user = s.query(User).filter_by(email="a@x.com").first() # 2.0 风格（推荐）：select() + Session.execute() from pbatis import select stmt = select(User).where(User.age > 18) for u in s.execute(stmt).scalars(): print(u.name) # 复杂条件 from pbatis import and_, or_, not_ stmt = select(User).where( and_( User.age > 18, or_(User.city == "深圳", User.city == "广州"), not_(User.name.like("test%")), ) ) ---- 数据库方言（Dialects） ---------------------- PyBatis 通过 URL 中的 ``dialect+driver://`` 形式选择数据库后端与驱动。 URL 格式 ~~~~~~~~ .. code-block:: text dialect+driver://user:password@host:port/database?param=value 参数说明： - ``dialect``：数据库类型（``sqlite``、``mysql``、``postgresql``、``oracle``、``mssql``） - ``driver``：驱动名称（小写，可选，省略时使用默认驱动） - 查询参数：作为连接参数透传给驱动 SQLite ~~~~~~ .. code-block:: python # 内存库（测试用） engine = create_engine("sqlite:///:memory:") # 内存共享（多连接可见） engine = create_engine("sqlite:///file:memdb1?mode=memory&cache=shared&uri=true") # 文件库 engine = create_engine("sqlite:///data/app.db") engine = create_engine("sqlite:////absolute/path/app.db") # Unix 绝对路径 # 异步（aiosqlite） engine = create_async_engine("sqlite+aiosqlite:///data/app.db") # 加密（SQLCipher，需安装 sqlcipher3-binary） engine = create_engine("sqlite+pysqlcipher://:password@/data/encrypted.db") # 特殊方言：pysqlite_numeric、pysqlite_dollar engine = create_engine("sqlite+pysqlite_numeric:///:memory:") # NUMERIC 精度修复 engine = create_engine("sqlite+pysqlite_dollar:///:memory:") # $ 字符串引用 MySQL / MariaDB ~~~~~~~~~~~~~~~ .. code-block:: python # PyMySQL（纯 Python，最常用） engine = create_engine("mysql+pymysql://user:pwd@localhost:3306/mydb?charset=utf8mb4") # mysqlclient（性能好，需 C 扩展） engine = create_engine("mysql+mysqldb://user:pwd@localhost:3306/mydb?charset=utf8mb4") # mysql-connector-python（Oracle 官方） engine = create_engine("mysql+mysqlconnector://user:pwd@localhost:3306/mydb") # MariaDB 官方驱动 engine = create_engine("mariadb+mariadbconnector://user:pwd@localhost:3306/mydb") # 异步 engine = create_async_engine("mysql+aiomysql://user:pwd@localhost:3306/mydb") engine = create_async_engine("mysql+asyncmy://user:pwd@localhost:3306/mydb") # 多主机（读写分离） engine = create_engine("mysql+pymysql://user:pwd@host1,host2/mydb") # 常用参数 # ?charset=utf8mb4 字符集 # ?ssl_ca=/path/ca.pem SSL 证书 # ?ssl_verify_cert=false 跳过证书验证 PostgreSQL ~~~~~~~~~~ .. code-block:: python # psycopg 3（推荐） engine = create_engine("postgresql+psycopg://user:pwd@localhost:5432/mydb") # psycopg2-binary engine = create_engine("postgresql+psycopg2://user:pwd@localhost:5432/mydb") # pg8000（纯 Python） engine = create_engine("postgresql+pg8000://user:pwd@localhost:5432/mydb") # 异步 engine = create_async_engine("postgresql+asyncpg://user:pwd@localhost:5432/mydb") # PG 特有：ARRAY / JSONB / HSTORE from pbatis.dialects.postgresql import ARRAY, JSONB, HSTORE from pbatis.dialects.postgresql import insert as pg_insert t = Table("t", m, Column("tags", ARRAY(String)), Column("attrs", JSONB)) # ON CONFLICT (UPSERT) stmt = pg_insert(t).values(id=1, name="x").on_conflict_do_update( index_elements=[t.c.id], set_={"name": "y"}, ) # JSONB 操作 from pbatis.dialects.postgresql import JSONB stmt = select(t).where(t.c.attrs["key"].astext == "value") stmt = select(t).where(t.c.attrs.contains({"k": "v"})) # 全文搜索 from pbatis.dialects.postgresql import TSVECTOR stmt = select(t).where(t.c.doc.op("@@")(func.to_tsquery("python & sql"))) Oracle ~~~~~~ .. code-block:: python # oracledb（推荐，替代旧的 cx_Oracle） engine = create_engine("oracle+oracledb://user:pwd@localhost:1521/?service_name=FREEPDB1") # cx_Oracle engine = create_engine("oracle+cx_oracle://user:pwd@localhost:1521/?service_name=FREEPDB1") # TNS 名称 engine = create_engine("oracle+oracledb://user:pwd@MY_TNS_ALIAS") # Oracle 特有类型 from pbatis.dialects.oracle import RAW, NUMBER, CLOB, BLOB, ROWID Microsoft SQL Server ~~~~~~~~~~~~~~~~~~~~ .. code-block:: python # pyodbc + ODBC Driver 17/18 engine = create_engine( "mssql+pyodbc://user:pwd@localhost:1433/mydb" "?driver=ODBC+Driver+18+for+SQL+Server" "&TrustServerCertificate=yes" ) # pymssql engine = create_engine("mssql+pymssql://user:pwd@localhost:1433/mydb") # mssql-python（微软官方纯 Python） engine = create_engine("mssql+mssqlpython://user:pwd@localhost:1433/mydb?Encrypt=No") # 异步 engine = create_async_engine("mssql+aioodbc://user:pwd@localhost:1433/mydb?driver=ODBC+Driver+18+for+SQL+Server") # MSSQL 特有：IDENTITY、ROWVERSION from pbatis.dialects.mssql import ROWVERSION t = Table("t", m, Column("id", Integer, primary_key=True), Column("version", ROWVERSION)) 方言能力矩阵 ~~~~~~~~~~~~ .. list-table:: :header-rows: 1 :widths: 15 12 12 12 12 12 12 * - 能力 - SQLite - MySQL - PostgreSQL - Oracle - MSSQL - SQLCipher * - 基础 DDL/DML - ✓ - ✓ - ✓ - ✓ - ✓ - ✓ * - RETURNING - ✓ - ✗ - ✓ - ✓ (21c+) - ✓ (OUTPUT) - ✓ * - UPSERT - ON CONFLICT - ON DUPLICATE KEY - ON CONFLICT - MERGE - MERGE - ON CONFLICT * - JSON 字段 - JSON - JSON - JSONB / JSON - CLOB+CHECK - NVARCHAR+ISJSON - JSON * - 数组 - ✗ - ✗ - ✓ - VARRAY - ✗ - ✗ * - 全文搜索 - FTS5 - FULLTEXT - tsvector - TextIndex - Full-Text - FTS5 * - 窗口函数 - ✓ - ✓ - ✓ - ✓ - ✓ - ✓ * - CTE - ✓ - ✓ - ✓ - ✓ - ✓ - ✓ * - 异步驱动 - aiosqlite - aiomysql / asyncmy - asyncpg - ✗ - aioodbc - aiosqlite ---- 扩展模块（PyBatis Ext） ------------------------ .. code-block:: python # declarative —— 声明式 ORM from pbatis.ext.declarative import declarative_base # automap —— 从已有数据库自动生成模型 from pbatis.ext.automap import automap_base Base = automap_base() Base.prepare(engine, reflect=True) # associationproxy —— 集合属性代理 from pbatis.ext.associationproxy import association_proxy class User(Base): keywords: Mapped[list[str]] = association_proxy("user_keywords", "value") # hybrid —— 同时在 Python 层与 SQL 层可用的属性 from pbatis.ext.hybrid import hybrid_property, hybrid_method class Product(Base): @hybrid_property def discounted_price(self): return self.price * 0.9 @discounted_price.expression def discounted_price(cls): return cls.price * 0.9 # mutable —— 可变类型变更跟踪 from pbatis.ext.mutable import MutableDict, MutableList class User(Base): preferences: Mapped[dict] = mapped_column(MutableDict.as_mutable(JSON)) # orderinglist —— 有序集合 from pbatis.ext.orderinglist import ordering_list class Playlist(Base): tracks: Mapped[list["Track"]] = relationship( "Track", order_by="Track.position", collection_class=ordering_list("position"), ) # serializer —— pickle 序列化大字段 from pbatis.ext.serializer import PickleType blobs: Mapped[dict] = mapped_column(PickleType) # horizontal_shard —— 横向分片 from pbatis.ext.horizontal_shard import ShardedSession # 通过 create_engine + 路由函数实现分库分表 # baked —— 预编译查询缓存 from pbatis.ext.baked import baked # indexable —— 支持 ``obj["key"]`` 访问的字典类型 from pbatis.ext.indexable import indexable # asyncio —— 异步 IO 支持 from pbatis.ext.asyncio import ( create_async_engine, AsyncSession, async_sessionmaker, ) ---- 扩展模块：雪花 ID（Snowflake） ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ .. code-block:: python # 导入 from pbatis.ext.snowflake import ( snowflake_id, # 便捷函数 SnowflakeID, # Column 类型 SnowflakeGenerator, # 完整生成器类 snowflake_column, # 工厂函数 configure_default, # 配置进程默认生成器 TWITTER_EPOCH, # 起始时间戳常量 ) ID 结构 ^^^^^^^^ 64 位二进制按 BigInteger 存储： .. list-table:: :header-rows: 1 :widths: 8 42 8 42 * - 位 - 含义 - 位 - 含义 * - 63 - 符号位（始终 0） - 22-12 - worker_id（5 bit，0-31） * - 62-22 - 时间戳（41 bit，相对 epoch 的毫秒数） - 11-0 - sequence（12 bit，每毫秒 4096 个） 默认 epoch = 1288834974657（Twitter 原版，2010-11-04 01:42:54 UTC）。 41 位时间戳可用约 69 年；5+5 位最多 1024 节点；12 位序列号单节点每毫秒 4096 个 ID。 集群 1024 节点理论峰值约每秒 40 亿 ID。 直接生成 ID ^^^^^^^^^^^^ .. code-block:: python from pbatis.ext.snowflake import snowflake_id, SnowflakeGenerator # 进程默认生成器 uid = snowflake_id() # 例如 1537200202186752 uid2 = snowflake_id() # 比 uid 大（趋势递增） # 自定义生成器 gen = SnowflakeGenerator(worker_id=1, datacenter_id=1) uid = gen.next_id() # 解析 ID（拆出时间戳、节点、序列号） info = gen.parse(uid) info["datetime"] # 生成时刻（UTC datetime） info["worker_id"] # 节点 ID info["datacenter_id"] info["sequence"] # 毫秒内序列号 info["timestamp"] # Unix 毫秒时间戳 用作主键（Core API） ^^^^^^^^^^^^^^^^^^^^ .. code-block:: python from pbatis import Table, Column, MetaData, String from pbatis.ext.snowflake import SnowflakeID, snowflake_id users = Table("users", MetaData(), Column("id", SnowflakeID, primary_key=True, default=snowflake_id), Column("name", String(50)), ) # INSERT 时自动生成主键 with engine.begin() as conn: r = conn.execute(users.insert().values(name="张三")) print(r.inserted_primary_key) # (2109665447048642560,) # 批量 INSERT 也会逐行生成不同 ID conn.execute(users.insert(), [ {"name": "李四"}, {"name": "王五"}, ]) # 也可指定专用生成器 from pbatis.ext.snowflake import SnowflakeGenerator custom_gen = SnowflakeGenerator(worker_id=2, datacenter_id=1) special = Table("special", MetaData(), Column("id", SnowflakeID, primary_key=True, default=custom_gen.next_id), Column("payload", String(100)), ) 工厂函数（更短） ^^^^^^^^^^^^^^^^ .. code-block:: python from pbatis.ext.snowflake import snowflake_column # 等价于 Column("id", SnowflakeID, primary_key=True, default=snowflake_id) t = Table("t", MetaData(), snowflake_column("id", primary_key=True), snowflake_column("user_id"), # 非主键也可用 Column("data", String(100)), ) ORM 用法 ^^^^^^^^ .. code-block:: python from pbatis.orm import Base, Mapped, mapped_column, Session from pbatis.ext.snowflake import SnowflakeID, snowflake_id class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(SnowflakeID, primary_key=True, default=snowflake_id) name: Mapped[str] with Session(engine) as s: u = User(name="赵六") s.add(u) s.commit() print(u.id) # 自动生成 进程级默认生成器 ^^^^^^^^^^^^^^^^ .. code-block:: python from pbatis.ext.snowflake import ( configure_default, get_default_generator, set_default_generator, ) # 应用启动时配置一次 import os configure_default( worker_id=int(os.environ.get("WORKER_ID", "0")), datacenter_id=int(os.environ.get("DC_ID", "0")), ) # 全局未指定 generator 的 SnowflakeID 都用此默认 from pbatis.ext.snowflake import snowflake_id uid = snowflake_id() # 使用默认生成器 # 运行时替换（不推荐） set_default_generator(SnowflakeGenerator(worker_id=5, datacenter_id=5)) 线程安全 ^^^^^^^^ 所有 ID 生成操作都通过内部锁保证线程安全；4 线程 × 100 次共 400 个 ID 全部唯一且递增。 .. code-block:: python import threading from pbatis.ext.snowflake import snowflake_id ids = [] def worker(): for _ in range(1000): ids.append(snowflake_id()) threads = [threading.Thread(target=worker) for _ in range(8)] for t in threads: t.start() for t in threads: t.join() assert len(set(ids)) == 8000 # 全部唯一 时钟回拨处理 ^^^^^^^^^^^^ .. code-block:: python # 容忍 ≤ 5ms 的回拨（自旋等待追上） # 超出则抛 ClockMovedBackwardsError from pbatis.ext.snowflake import SnowflakeGenerator, ClockMovedBackwardsError gen = SnowflakeGenerator(worker_id=0, datacenter_id=0) try: gen.next_id() except ClockMovedBackwardsError as e: # 时钟回拨超过 5ms；记录告警 + 等待 + 重试 logger.error("Clock skew detected: %s", e) time.sleep(0.1) gen.next_id() ID 业务化（带业务前缀） ~~~~~~~~~~~~~~~~~~~~~~~~~ 如需带业务前缀（如 ``order_``、``user_``），不要直接修改 SnowflakeGenerator，而是封装一个字段： .. code-block:: python from pbatis import Table, Column, MetaData, String from pbatis.ext.snowflake import SnowflakeID, snowflake_id # 主表存原始 bigint 雪花 ID orders = Table("orders", MetaData(), Column("id", SnowflakeID, primary_key=True, default=snowflake_id), Column("biz_id", String(30), default=lambda: f"ord_{snowflake_id()}"), Column("amount", String(20)), ) ---- 配置与连接池 ------------ .. code-block:: python # 调试：打印 SQL engine = create_engine("postgresql+psycopg://...", echo=True) engine.echo = True # 运行时切换 # 只打印部分 import logging logging.basicConfig() logging.getLogger("pbatis.engine").setLevel(logging.INFO) # 池配置 engine = create_engine( "postgresql+psycopg://...", pool_size=10, max_overflow=20, pool_timeout=30, # 取连接等待秒数 pool_recycle=3600, # 多久回收连接 pool_pre_ping=True, # 取出时验证连接有效 ) # 关闭连接池（每次新建） engine = create_engine("sqlite:///:memory:", poolclass=NullPool) # 静态池（所有连接共享同一物理连接，适合内存 SQLite） engine = create_engine("sqlite:///:memory:", poolclass=StaticPool, connect_args={"check_same_thread": False}) # 异步连接池 engine = create_async_engine( "postgresql+asyncpg://...", pool_size=5, max_overflow=10, ) # 关闭引擎 engine.dispose() 事件系统 -------- .. code-block:: python from pbatis import event # 监听连接建立 @event.listens_for(engine, "connect") def receive_connect(dbapi_connection, connection_record): cursor = dbapi_connection.cursor() cursor.execute("SET time_zone = '+08:00'") # 监听连接 checkout（取出） @event.listens_for(engine, "checkout") def receive_checkout(dbapi_connection, connection_record, connection_proxy): pass # 监听 ORM 事件 from pbatis.orm import Session @event.listens_for(Session, "before_commit") def before_commit(session): pass @event.listens_for(User, "load") def receive_load(target, context): pass ---- 类型与自定义 ------------ .. code-block:: python from pbatis import TypeDecorator, String # 自定义类型：加密字段 class EncryptedString(TypeDecorator): impl = String cache_ok = True def process_bind_param(self, value, dialect): return encrypt(value) if value else value def process_result_value(self, value, dialect): return decrypt(value) if value else value # 自定义类型：JSON 在 SQLite/MySQL 走 TEXT，在 PG 走 JSONB class JsonField(TypeDecorator): impl = String cache_ok = True def load_dialect_impl(self, dialect): if dialect.name == "postgresql": from pbatis.dialects.postgresql import JSONB return dialect.type_descriptor(JSONB()) return dialect.type_descriptor(String()) # 枚举字段 import enum class Color(enum.Enum): RED = "red" GREEN = "green" BLUE = "blue" Column("color", Enum(Color)) ---- 反射（Reflection） ------------------ 从已有数据库读取表结构，无需手写 Table 定义： .. code-block:: python from pbatis import MetaData, Table, create_engine, inspect engine = create_engine("postgresql+psycopg://localhost/mydb") metadata = MetaData() # 反射单张表 users = Table("users", metadata, autoload_with=engine) print(users.c) # 反射所有表 metadata.reflect(bind=engine) for table in metadata.sorted_tables: print(table.name) # 用 inspect API inspector = inspect(engine) print(inspector.get_table_names()) print(inspector.get_columns("users")) print(inspector.get_foreign_keys("orders")) print(inspector.get_indexes("users")) print(inspector.get_pk_constraint("users")) print(inspector.get_unique_constraints("users")) print(inspector.get_check_constraints("users")) ---- 迁移与版本管理 -------------- PyBatis 不内置 schema 迁移工具；推荐使用外部迁移工具（业内常用 `Alembic`）配合 PyBatis： .. code-block:: bash pip install alembic alembic init migrations # 在 alembic/env.py 中将 target_metadata 指向 Base.metadata ---- 使用示例 -------- 以下 4 个示例覆盖最常见的使用场景，全部可独立运行。 完整博客系统（端到端） ~~~~~~~~~~~~~~~~~~~~~~ 一个含用户、文章、评论、标签（多对多）的完整博客系统，从建表到查询一气呵成。 .. code-block:: python from datetime import datetime from pbatis import ( create_engine, MetaData, Table, Column, Integer, String, Text, ForeignKey, DateTime, Boolean, select, insert, update, delete, func, and_, or_, ) from pbatis.orm import ( declarative_base, Mapped, mapped_column, relationship, Session, sessionmaker, selectinload, ) # ---------- 1. Schema 定义（Core API）---------- metadata = MetaData() users = Table("users", metadata, Column("id", Integer, primary_key=True, autoincrement=True), Column("username", String(50), unique=True, nullable=False), Column("email", String(120), unique=True, nullable=False), Column("password_hash", String(128), nullable=False), Column("bio", Text), Column("created_at", DateTime, server_default=func.now()), ) posts = Table("posts", metadata, Column("id", Integer, primary_key=True, autoincrement=True), Column("author_id", Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False), Column("title", String(200), nullable=False), Column("body", Text, nullable=False), Column("published", Boolean, default=False), Column("view_count", Integer, default=0), Column("created_at", DateTime, server_default=func.now()), Column("updated_at", DateTime, onupdate=func.now()), ) comments = Table("comments", metadata, Column("id", Integer, primary_key=True, autoincrement=True), Column("post_id", Integer, ForeignKey("posts.id", ondelete="CASCADE"), nullable=False), Column("user_id", Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False), Column("content", Text, nullable=False), Column("created_at", DateTime, server_default=func.now()), ) # 关联表：多对多（文章 <-> 标签） post_tags = Table("post_tags", metadata, Column("post_id", Integer, ForeignKey("posts.id", ondelete="CASCADE"), primary_key=True), Column("tag_id", Integer, ForeignKey("tags.id", ondelete="CASCADE"), primary_key=True), ) tags = Table("tags", metadata, Column("id", Integer, primary_key=True, autoincrement=True), Column("name", String(50), unique=True, nullable=False), Column("slug", String(60), unique=True, nullable=False), ) # ---------- 2. 引擎与会话工厂 ---------- engine = create_engine("sqlite:///:memory:", echo=False) metadata.create_all(engine) SessionLocal = sessionmaker(engine, expire_on_commit=False) # ---------- 3. 增 ---------- with SessionLocal() as s: # 用户 s.execute(insert(users).values( username="alice", email="alice@x.com", password_hash="hash1", bio="pythonista" )) s.execute(insert(users).values( username="bob", email="bob@x.com", password_hash="hash2" )) # 文章 s.execute(insert(posts).values( author_id=1, title="PyBatis 入门", body="PyBatis 是一个 Python ORM 库", published=True, view_count=100 )) s.execute(insert(posts).values( author_id=1, title="高级用法", body="事务、锁、连接池", published=False, view_count=10 )) # 标签 + 多对多 s.execute(insert(tags).values(name="Python", slug="python")) s.execute(insert(tags).values(name="ORM", slug="orm")) s.execute(insert(post_tags).values(post_id=1, tag_id=1)) s.execute(insert(post_tags).values(post_id=1, tag_id=2)) s.execute(insert(post_tags).values(post_id=2, tag_id=1)) # 评论 s.execute(insert(comments).values( post_id=1, user_id=2, content="好文！" )) s.commit() # ---------- 4. 查 ---------- with SessionLocal() as s: # 全部已发布文章 stmt = select(posts).where(posts.c.published == True).order_by(posts.c.view_count.desc()) for row in s.execute(stmt): print(f" [{row.id}] {row.title} (浏览 {row.view_count})") # 某用户全部文章 + 评论数 stmt = ( select( posts, func.count(comments.c.id).label("comment_cnt"), ) .outerjoin(comments, comments.c.post_id == posts.c.id) .where(posts.c.author_id == 1) .group_by(posts.c.id) .order_by(posts.c.id) ) for row in s.execute(stmt): print(f" [{row.id}] {row.title} - {row.comment_cnt} 条评论") # 标签为 Python 的文章 stmt = ( select(posts) .join(post_tags, post_tags.c.post_id == posts.c.id) .join(tags, tags.c.id == post_tags.c.tag_id) .where(tags.c.slug == "python") ) for row in s.execute(stmt): print(f" Tag 'python': [{row.id}] {row.title}") # ---------- 5. 改 ---------- with SessionLocal() as s: # 发布某文章 + 浏览数 +1 s.execute( update(posts) .where(posts.c.id == 2) .values(published=True, view_count=posts.c.view_count + 1) ) s.commit() # ---------- 6. 删 ---------- with SessionLocal() as s: # 删除某评论 s.execute(delete(comments).where(comments.c.id == 1)) s.commit() # ---------- 7. ORM 模型版（更易维护）---------- Base = declarative_base() class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(primary_key=True) username: Mapped[str] = mapped_column(String(50), unique=True) email: Mapped[str] = mapped_column(String(120), unique=True) posts: Mapped[list["Post"]] = relationship( "Post", back_populates="author", cascade="all, delete-orphan" ) comments: Mapped[list["Comment"]] = relationship( "Comment", back_populates="user", cascade="all, delete-orphan" ) class Post(Base): __tablename__ = "posts" id: Mapped[int] = mapped_column(primary_key=True) author_id: Mapped[int] = mapped_column(ForeignKey("users.id")) title: Mapped[str] = mapped_column(String(200)) body: Mapped[str] = mapped_column(Text) published: Mapped[bool] = mapped_column(default=False) view_count: Mapped[int] = mapped_column(default=0) author: Mapped["User"] = relationship("User", back_populates="posts") comments: Mapped[list["Comment"]] = relationship( "Comment", back_populates="post", cascade="all, delete-orphan" ) tags: Mapped[list["Tag"]] = relationship( "Tag", secondary="post_tags", back_populates="posts" ) class Comment(Base): __tablename__ = "comments" id: Mapped[int] = mapped_column(primary_key=True) post_id: Mapped[int] = mapped_column(ForeignKey("posts.id")) user_id: Mapped[int] = mapped_column(ForeignKey("users.id")) content: Mapped[str] = mapped_column(Text) post: Mapped["Post"] = relationship("Post", back_populates="comments") user: Mapped["User"] = relationship("User", back_populates="comments") class Tag(Base): __tablename__ = "tags" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] = mapped_column(String(50), unique=True) slug: Mapped[str] = mapped_column(String(60), unique=True) posts: Mapped[list["Post"]] = relationship( "Post", secondary="post_tags", back_populates="tags" ) # ---------- 8. ORM 用法（推荐）---------- with SessionLocal() as s: # 查"已发布"文章（预加载作者、标签、评论） stmt = ( select(Post) .options(selectinload(Post.author), selectinload(Post.tags)) .where(Post.published == True) .order_by(Post.view_count.desc()) ) for post in s.execute(stmt).scalars(): print(f" {post.title} by {post.author.username}, 标签={[t.name for t in post.tags]}") # 增 new_post = Post( author_id=2, title="ORM 体验", body="...", tags=[s.query(Tag).filter_by(slug="orm").one()] ) s.add(new_post) s.commit() # 改 post = s.query(Post).filter_by(title="ORM 体验").one() post.view_count += 1 s.commit() # 删 s.query(Comment).filter_by(id=1).delete() s.commit() FastAPI 集成 ~~~~~~~~~~~~ PyBatis 2.0 + FastAPI 的标准集成模式：用 Depends 注入 Session。 .. code-block:: python # main.py from typing import Generator from fastapi import Depends, FastAPI, HTTPException from pbatis import create_engine from pbatis.orm import Session, sessionmaker from pbatis.orm import declarative_base, Mapped, mapped_column, Integer, String Base = declarative_base() engine = create_engine("postgresql+psycopg://user:pwd@localhost/mydb") SessionLocal = sessionmaker(engine, expire_on_commit=False, autoflush=False) class Item(Base): __tablename__ = "items" id: Mapped[int] = mapped_column(Integer, primary_key=True) name: Mapped[str] = mapped_column(String(100)) price: Mapped[float] = mapped_column(default=0) Base.metadata.create_all(engine) app = FastAPI() # 依赖：每个请求一个 Session def get_db() -> Generator[Session, None, None]: db = SessionLocal() try: yield db finally: db.close() # CRUD 端点 @app.post("/items") def create_item(name: str, price: float, db: Session = Depends(get_db)): item = Item(name=name, price=price) db.add(item) db.commit() db.refresh(item) return {"id": item.id, "name": item.name} @app.get("/items/{item_id}") def get_item(item_id: int, db: Session = Depends(get_db)): item = db.get(Item, item_id) if not item: raise HTTPException(404, "item not found") return {"id": item.id, "name": item.name, "price": item.price} @app.get("/items") def list_items(skip: int = 0, limit: int = 20, db: Session = Depends(get_db)): from pbatis import select stmt = select(Item).offset(skip).limit(limit) return [{"id": i.id, "name": i.name, "price": i.price} for i in db.execute(stmt).scalars()] @app.put("/items/{item_id}") def update_item(item_id: int, name: str, price: float, db: Session = Depends(get_db)): item = db.get(Item, item_id) if not item: raise HTTPException(404) item.name = name item.price = price db.commit() return {"id": item.id} @app.delete("/items/{item_id}") def delete_item(item_id: int, db: Session = Depends(get_db)): item = db.get(Item, item_id) if not item: raise HTTPException(404) db.delete(item) db.commit() return {"deleted": item_id} # 异步版本（与 FastAPI 异步端点配合） from pbatis.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession async_engine = create_async_engine("postgresql+asyncpg://user:pwd@localhost/mydb") AsyncSessionLocal = async_sessionmaker(async_engine, expire_on_commit=False) async def get_async_db() -> AsyncSession: async with AsyncSessionLocal() as s: yield s @app.get("/async-items/{item_id}") async def get_async_item(item_id: int, db: AsyncSession = Depends(get_async_db)): from pbatis import select stmt = select(Item).where(Item.id == item_id) result = await db.execute(stmt) item = result.scalar_one_or_none() if not item: raise HTTPException(404) return {"id": item.id, "name": item.name} Flask 集成 ~~~~~~~~~~ .. code-block:: python # app.py from flask import Flask, g from pbatis import create_engine from pbatis.orm import sessionmaker, Session app = Flask(__name__) engine = create_engine("postgresql+psycopg://user:pwd@localhost/mydb") SessionLocal = sessionmaker(engine) @app.before_request def create_session(): g.db = SessionLocal() @app.teardown_request def close_session(exception=None): db = g.pop("db", None) if db is not None: db.close() @app.route("/items") def list_items(): from pbatis import select from myapp.models import Item # 你的 ORM 模型 items = g.db.execute(select(Item).limit(20)).scalars().all() return {"items": [{"id": i.id, "name": i.name} for i in items]} @app.route("/items/", methods=["DELETE"]) def delete_item(item_id): from myapp.models import Item item = g.db.get(Item, item_id) if not item: return {"error": "not found"}, 404 g.db.delete(item) g.db.commit() return {"deleted": item_id} Django 集成 ~~~~~~~~~~~ PyBatis 不替换 Django ORM；可作为外部 SQL 工具处理复杂查询或迁移到 Django 之外的存储。 .. code-block:: python # 在 Django settings.py 配置 DATABASES = { "default": { "ENGINE": "django.db.backends.postgresql", "NAME": "mydb", "USER": "user", "PASSWORD": "pwd", "HOST": "localhost", } } # myapp/db.py —— PyBatis 引擎单例 import os from pbatis import create_engine from pbatis.orm import sessionmaker, declarative_base DB_URL = "postgresql+psycopg://{USER}:{PASSWORD}@{HOST}/{NAME}".format(**os.environ) engine = create_engine(DB_URL, pool_pre_ping=True) SessionLocal = sessionmaker(engine) Base = declarative_base() # myapp/views.py from django.http import JsonResponse from myapp.db import SessionLocal from myapp.pyb_models import Order # PyBatis 定义的模型 def order_list(request): from pbatis import select with SessionLocal() as s: orders = s.execute(select(Order).limit(50)).scalars().all() return JsonResponse({"orders": [o.to_dict() for o in orders]}, safe=False) 生产部署清单 ~~~~~~~~~~~~ .. code-block:: python # 1. 创建引擎（生产配置） engine = create_engine( os.environ["DATABASE_URL"], pool_size=20, # 池大小 max_overflow=10, # 突发容量 pool_recycle=3600, # 1 小时回收 pool_pre_ping=True, # 取连接前 ping pool_timeout=30, # 等待超时 ) # 2. 进程退出时关闭引擎 import atexit atexit.register(engine.dispose) # 3. 优雅关闭时 dispose（如 gunicorn worker exit） def on_worker_exit(server, worker): engine.dispose() # 4. 健康检查 from pbatis import text def health_check() -> bool: try: with engine.connect() as conn: conn.execute(text("SELECT 1")) return True except Exception: return False # 5. 多租户（按请求选引擎） from typing import Dict from pbatis import Engine, create_engine _engines: Dict[str, Engine] = {} def get_engine(tenant_id: str) -> Engine: if tenant_id not in _engines: _engines[tenant_id] = create_engine( f"postgresql+psycopg://user:pwd@host/tenant_{tenant_id}", pool_size=5, ) return _engines[tenant_id] 测试用例 ~~~~~~~~ .. code-block:: python # 用 pytest + 内存 SQLite 跑 PyBatis 应用测试 # conftest.py import pytest from pbatis import create_engine from pbatis.orm import sessionmaker from myapp.models import Base @pytest.fixture def engine(): eng = create_engine("sqlite:///:memory:") Base.metadata.create_all(eng) yield eng eng.dispose() @pytest.fixture def session(engine): SessionLocal = sessionmaker(engine) s = SessionLocal() yield s s.close() # test_users.py from pbatis import select from myapp.models import User def test_create_user(session): u = User(username="alice", email="a@x.com") session.add(u) session.commit() assert u.id is not None def test_query_user(session): u = session.execute(select(User).where(User.username == "alice")).scalar_one() assert u.email == "a@x.com" ---- 性能调优 -------- 从连接池、查询、ORM、批量操作、流式读取 5 个层面系统讲解如何榨干 PyBatis 性能。 启用 Cython 扩展 ~~~~~~~~~~~~~~~~ 最快的速度提升：编译 C 扩展，常见操作（行构造、类型转换、缓存键、结果集迭代）从纯 Python 切换到 C 实现，平均 30%-200% 提升。 .. code-block:: bash pip install cython>=3.3 pip install -e . # 自动编译 .pyx 编译产物在 ``lib/pbatis/engine/_processors_cy.cpython-*.so``、``_row_cy.so``、``_result_cy.so``、``_cache_key_cy.so``、``_util_cy.so``、``_collections_cy.so``、``_immutabledict_cy.so``。 .. code-block:: python # 运行时检查 C 扩展是否启用 import pbatis print(pbatis.__version__) import pbatis.engine._processors_cy as _ print("C extension: enabled") # 没编译会 ImportError 连接池调优 ~~~~~~~~~~ .. code-block:: python # 1. 池大小 # 经验值：CPU * 2-4 倍（IO 密集型可更大） engine = create_engine( "postgresql+psycopg://...", pool_size=20, max_overflow=10, # 突发容量 pool_timeout=30, # 取连接等待秒 ) # 2. 回收策略 engine = create_engine( "postgresql+psycopg://...", pool_recycle=3600, # 1 小时回收（避免数据库主动断开） pool_pre_ping=True, # 取连接前 ping（防 stale 连接） ) # 3. 不同场景的池 # 单进程单请求测试：NullPool engine = create_engine(url, poolclass=NullPool) # 内存 SQLite：StaticPool 共享同一连接 engine = create_engine( "sqlite:///:memory:", poolclass=StaticPool, connect_args={"check_same_thread": False}, ) # 异步引擎池 from pbatis.ext.asyncio import create_async_engine eng = create_async_engine("postgresql+asyncpg://...", pool_size=10, max_overflow=20) 避免 N+1 查询 ~~~~~~~~~~~~~ N+1 是 ORM 最常见的性能陷阱：1 条主查询 + N 条子查询。 .. code-block:: python # 错：触发 N+1（10 个 author → 11 条 SQL） posts = s.execute(select(Post)).scalars().all() for p in posts: print(p.author.username) # 每条都触发一次 SELECT # 对 1：selectinload（推荐，1 条额外 IN 查询） from pbatis.orm import selectinload stmt = select(Post).options(selectinload(Post.author)) posts = s.execute(stmt).scalars().all() for p in posts: print(p.author.username) # 共 2 条 SQL # 对 2：joinedload（1 条 LEFT JOIN） from pbatis.orm import joinedload stmt = select(Post).options(joinedload(Post.author)) # 对 3：subqueryload（1 条子查询） from pbatis.orm import subqueryload stmt = select(Post).options(subqueryload(Post.author)) # 多层预加载 stmt = select(User).options( selectinload(User.posts).selectinload(Post.tags), selectinload(User.comments), ) # 手写 JOIN 但仍按 relationship 填充 from pbatis.orm import contains_eager stmt = ( select(User) .join(Post) .where(Post.published == True) .options(contains_eager(User.posts)) ) 只取需要的列 ~~~~~~~~~~~~ .. code-block:: python # 错：取全部列（包括大字段） stmt = select(User) # 包含 bio（Text）、avatar（LargeBinary）等 # 对：只取需要的列 stmt = select(User.id, User.username, User.email) # ORM 也支持列级限定 from pbatis.orm import defer, deferrable, undefer stmt = select(User).options(defer(User.bio)) # 默认不取 bio s.execute(stmt).scalar().bio # 访问时才触发加载 # 一次性取消 defer s.execute(stmt).scalars().all() s.execute(select(User).options(undefer(User.bio))).all() # 这次取 批量操作 ~~~~~~~~ INSERT / UPDATE 逐条提交是最常见的慢来源之一。PyBatis 提供 3 个层级的批量能力。 .. code-block:: python # 1. Core API executemany（最快，自动用驱动原生批处理） with engine.begin() as conn: conn.execute(insert(users), [ {"name": "u1"}, {"name": "u2"}, {"name": "u3"}, # ... 10000 行 ]) # 2. Session.bulk_save_objects（ORM 批量 INSERT） session.bulk_save_objects([ User(username=f"user_{i}") for i in range(10000) ]) session.commit() # 3. Session.bulk_insert_mappings（指定列） session.bulk_insert_mappings( User, [{"username": f"u{i}", "email": f"u{i}@x.com"} for i in range(10000)], ) session.commit() # 4. Session.bulk_update_mappings（按主键批量 UPDATE） session.bulk_update_mappings(User, [ {"id": 1, "bio": "new bio 1"}, {"id": 2, "bio": "new bio 2"}, ]) session.commit() # 性能对比（10 万行 INSERT 到 PG，psycopg） # 逐条 + commit: 47 秒 # executemany: 1.8 秒 # COPY (raw): 0.4 秒（最快但绕过 ORM） # 5. PG 的最快方式：用 raw psycopg.copy_from import io from pbatis import text with engine.raw_connection() as raw: with raw.cursor() as cur: buf = io.StringIO() for u in users_data: buf.write(f"{u['name']}\t{u['email']}\n") buf.seek(0) cur.copy_from(buf, "users", columns=("name", "email")) raw.commit() 流式读取大数据集 ~~~~~~~~~~~~~~~~ 处理百万行结果时不要一次性加载到内存。 .. code-block:: python # 1. Core API 流式 with engine.connect() as conn: result = conn.execution_options(stream_results=True).execute( select(events).order_by(events.c.id) ) for row in result: process(row) # 一次一行 # 2. yield_per（ORM 批量缓冲） with Session(engine) as s: stmt = select(Event).execution_options(yield_per=1000) for event in s.execute(stmt).scalars(): process(event) # yield_per 1000 意味着每 1000 行从游标取一次 # 3. 仅取主键再分批 with Session(engine) as s: total = s.execute(select(func.count(Event.id))).scalar() for offset in range(0, total, 5000): batch = s.execute( select(Event).where(Event.id >= offset).limit(5000) ).scalars().all() process_batch(batch) Session 配置 ~~~~~~~~~~~~ .. code-block:: python # 1. expire_on_commit=False（重要） # 默认 True：commit 后实例属性过期，再次访问会重新查询 # 设 False：commit 后属性仍可用，性能更好 SessionLocal = sessionmaker(engine, expire_on_commit=False) # 2. autoflush=False（批量导入时） # 默认 True：每次查询前自动 flush pending 对象 # 批量导入时关闭可避免无谓的 flush session = SessionLocal(autoflush=False) session.bulk_save_objects([User(username=f"u{i}") for i in range(1000)]) session.commit() # 3. autobegin=False（显式控制事务边界） # 不常用 # 4. 关闭 identity map（不缓存对象） session = SessionLocal() # 暂未提供原生开关，可改用 bulk_* 系列方法 # 5. 查询后再 expire 全部对象（释放内存） session.expire_all() # 6. detach 单个对象 session.expunge(user) 事务粒度 ~~~~~~~~ .. code-block:: python # 错：每条 INSERT 一个事务（10000 次 commit = 慢） for u in users_data: with engine.begin() as conn: conn.execute(insert(users).values(**u)) # 对：1 个事务包 10000 条 with engine.begin() as conn: for u in users_data: conn.execute(insert(users).values(**u)) # 更好：executemany with engine.begin() as conn: conn.execute(insert(users), users_data) # 长事务 vs 短事务 # 经验：单个事务 1-10 秒；长事务会持有锁、阻塞 vacuum # 批处理：每 1000 行 1 个事务（折中） BATCH = 1000 with engine.begin() as conn: for i, u in enumerate(users_data): conn.execute(insert(users).values(**u)) if (i + 1) % BATCH == 0: # 这里不能用嵌套 begin（savepoint 模式） pass # 批量提交（多次短事务） for i in range(0, len(users_data), BATCH): batch = users_data[i:i + BATCH] with engine.begin() as conn: conn.execute(insert(users), batch) SQL 层优化 ~~~~~~~~~~ .. code-block:: python # 1. 索引友好查询 # 前缀 % 走不到索引 → 全文搜索替代 select(users).where(users.c.name.like("张%")) # 走索引 select(users).where(users.c.name.like("%张")) # 全表扫描 # 用 FULLTEXT 替代：tsvector、MySQL MATCH、SQLite FTS5 # 2. EXISTS 替代 IN（大子查询） select(users).where(users.c.id.in_(select(orders.c.user_id))) # 可能慢 select(users).where(exists().where(orders.c.user_id == users.c.id)) # 短路 # 3. UNION ALL 替代 UNION（不排序去重） from pbatis import union_all union_all(...) # 更快 # 4. 避免 SELECT * select(users.c.id, users.c.name) # OK select(users) # 全列 # 5. 配合覆盖索引 # 如果查询只需要 (a, b)，且有索引 idx_c(a, b, c)，则索引扫描即可 select(t.c.a, t.c.b) where(t.c.a == 1) # 走覆盖索引，无须回表 # 6. 用 EXPLAIN 分析 from pbatis import text plan = s.execute(text("EXPLAIN ANALYZE SELECT * FROM users WHERE name = :n"), {"n": "alice"}).fetchall() for row in plan: print(row[0]) 异步 + 并发 ~~~~~~~~~~~ .. code-block:: python # 1. 异步批量并发 import asyncio from pbatis.ext.asyncio import create_async_engine, AsyncSession async def fetch_user(s, uid): from pbatis import select result = await s.execute(select(User).where(User.id == uid)) return result.scalar_one_or_none() async def main(): eng = create_async_engine("postgresql+asyncpg://...") async with eng.begin() as conn: await conn.run_sync(Base.metadata.create_all) async with AsyncSession(eng) as s: # 并发查 100 个用户 users = await asyncio.gather(*[fetch_user(s, i) for i in range(1, 101)]) return users # 2. asyncmy 配合 PyMySQL 池 eng = create_async_engine("mysql+asyncmy://...") # 3. asyncpg（PG 最快的异步驱动） eng = create_async_engine("postgresql+asyncpg://...") 诊断与监控 ~~~~~~~~~~ .. code-block:: python # 1. echo SQL engine = create_engine(url, echo=True) # 全部 SQL engine.echo = False # 运行时关闭 # 2. 部分 SQL 日志 import logging logging.basicConfig() logging.getLogger("pbatis.engine").setLevel(logging.INFO) logging.getLogger("pbatis.pool").setLevel(logging.DEBUG) # 连接池事件 logging.getLogger("pbatis.orm").setLevel(logging.INFO) # ORM 事件 # 3. 慢查询捕获（自定义事件） from pbatis import event import time @event.listens_for(engine, "before_execute") def before(conn, clauseelement, multiparams, params, execution_options): conn._query_start = time.time() @event.listens_for(engine, "after_execute") def after(conn, clauseelement, multiparams, params, result, execution_options): elapsed = time.time() - conn._query_start if elapsed > 0.5: logger.warning(f"Slow query ({elapsed:.2f}s): {clauseelement}") # 4. 监控：每条 SQL 计数 from collections import Counter counter = Counter() @event.listens_for(engine, "after_execute") def count(conn, clauseelement, *a, **kw): table = clauseelement.table.name if hasattr(clauseelement, "table") else "?" counter[table] += 1 # 查最热表 print(counter.most_common(10)) # 5. pg_stat_statements（PG） s.execute(text("SELECT * FROM pg_stat_statements ORDER BY total_time DESC LIMIT 10")) 常见性能陷阱 ~~~~~~~~~~~~ .. list-table:: :header-rows: 1 :widths: 45 55 * - 陷阱 - 修法 * - 触发 N+1（懒加载关联对象） - 用 ``selectinload`` / ``joinedload`` 预加载 * - ``session.query().all()`` 一次加载全表 - 用 ``yield_per`` 流式 + 批处理 * - ``expire_on_commit=True`` 重复查询 - 设 ``expire_on_commit=False`` * - 逐条 INSERT + 逐条 commit - 1 个事务 + executemany * - ``SELECT *`` 拉大字段 - 显式列出列；``defer`` 不常用大字段 * - 前缀 ``%`` LIKE 用不到索引 - 用全文搜索 / 反向索引 * - 每次 query 触发 autoflush - 批量导入时 ``autoflush=False`` * - 长事务持有锁 - 拆分多个短事务（每 1k 行一个） * - ``session.merge()`` 全字段对比 - 用 ``bulk_update_mappings`` 显式列 * - 反射时 ``autoload=True`` 启动慢 - 用 ``reflect()`` 一次性反射 性能基线（参考） ~~~~~~~~~~~~~~~~ 在不同场景下，PyBatis 2.x + Cython 大致能达到： .. list-table:: :header-rows: 1 :widths: 30 22 24 24 * - 操作 - SQLite (内存) - PostgreSQL (psycopg) - MySQL (PyMySQL) * - 单条 SELECT（已建索引） - ~50 µs - ~200 µs - ~300 µs * - 单条 INSERT - ~80 µs - ~300 µs - ~500 µs * - 批量 INSERT（1 万行，executemany） - ~50 ms - ~250 ms - ~500 ms * - N+1 查询（10 个子对象） - 11 次 SQL → 1 次（用 selectinload 后） - 同左 - 同左 * - ORM 单对象 fetch（懒加载） - ~60 µs - ~250 µs - ~350 µs *数据为 2024 年中高端笔记本（Apple M2 Pro / 16GB / 本地 PG 14）的参考值。生产环境因网络、并发、索引设计差异可能 ±50%。* ---- 常见问题 -------- **Q: ``import pbatis`` 报 ModuleNotFoundError？** 确认已通过 ``pip install -e .`` 或 ``pip install pbatis`` 安装；或在脚本中手动 ``sys.path.insert(0, "lib")``。 **Q: SQLite 文件被占用？** 默认 SQLite 仅允许单线程访问同一连接。多线程需 ``connect_args={"check_same_thread": False}`` 或使用 StaticPool。 **Q: 如何处理 Pymysql / 异步驱动的警告？** 设置 ``pbatis_warn_20 = "false"`` 或在环境变量中 ``export PBATIS_WARN_20=false``。 **Q: 性能调优要点？** 1. 用 ``engine.execution_options(stream_results=True)`` 流式读取大数据 2. 批量 ``insert()`` / ``executemany`` 3. ORM 懒加载关掉、用 ``selectinload`` / ``joinedload`` 预加载 4. 用 ``session.execute()`` + 2.0 风格 select()，避免 1.x Query 缓存开销 5. 启用 Cython 加速：``pip install cython`` 然后 ``pip install -e .`` 时会自动编译 ``_processors_cy``、``_row_cy``、``_result_cy``、``_cache_key_cy``、``_util_cy``、``_collections_cy``、``_immutabledict_cy`` 等 C 扩展 **Q: 如何关闭所有 SQL 日志？** .. code-block:: python import logging logging.getLogger("pbatis.engine").setLevel(logging.WARNING) **Q: 异步与同步能否混用？** PyBatis 的同步 Engine 和 AsyncEngine 互不通用，需分别创建。同一进程内可同时存在两种 Session（绑定不同 Engine）。 ---- 版本与许可 ---------- 当前版本：1.1.0 PyPI 包名：**pbatis**；安装命令 ``pip install pbatis``，项目页 https://pypi.org/project/pbatis/ 更多资源： - 完整变更日志：``CHANGELOG.md`` - 一页纸速查：``doc/build/quickref_zh.rst`` - 源码仓库：https://gitee.com/pbatis/pbatis 协议：MIT（详见 ``LICENSE`` 文件）