Skip to main content

PyBatis

Python 数据库工具与对象关系映射(ORM)框架。

提供 SQL 表达式语言、连接池、Schema 元数据、声明式 ORM 和跨方言的数据库访问能力,支持同步与异步两种编程模型,覆盖 SQLite、MySQL/MariaDB、PostgreSQL、Oracle、Microsoft SQL Server 等主流关系型数据库。


安装

pip install pybatis

按需安装特定方言驱动:

# SQLite(标准库自带 pysqlite,无需额外安装)
pip install pybatis[aiosqlite]        # 异步 SQLite
pip install pybatis[sqlcipher]        # 加密 SQLite

# MySQL / MariaDB
pip install pybatis[mysql]            # mysqlclient
pip install pybatis[pymysql]          # PyMySQL(纯 Python)
pip install pybatis[mysql-connector]  # mysql-connector-python
pip install pybatis[mariadb-connector]# mariadb 官方驱动
pip install pybatis[aiomysql]
pip install pybatis[asyncmy]

# PostgreSQL
pip install pybatis[postgresql]              # psycopg 3
pip install pybatis[postgresql-psycopg2binary]  # psycopg2-binary
pip install pybatis[postgresql-pg8000]
pip install pybatis[postgresql-asyncpg]

# Oracle
pip install pybatis[oracle]            # oracledb(推荐)
pip install pybatis[oracle-cxoracle]   # cx_Oracle

# Microsoft SQL Server
pip install pybatis[mssql]             # pyodbc
pip install pybatis[mssql-pymssql]
pip install pybatis[mssql-python]      # mssql-python

# 异步支持(greenlet)
pip install pybatis[asyncio]

要求 Python 3.11 及以上。


快速开始

from pybatis 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 访问数据库的入口,内部维护一个连接池。

from pybatis 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 提供,支持上下文管理器自动释放。

# 自动提交(事务)
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 pybatis.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() 声明。

from pybatis 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 pybatis 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)),
)

数据类型速查

类别

类型

适用场景

整数

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)

from pybatis 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 pybatis.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 pybatis.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 pybatis 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)

from pybatis 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)

from pybatis 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 pybatis 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)

from pybatis 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 pybatis 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 pybatis 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 子句里能写什么——一次性讲清。

比较运算符

from pybatis 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 pybatis 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)
)

运算符速查

运算符

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

from pybatis 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 pybatis 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 pybatis import any_, all_
select(products).where(products.c.price < any_(
    select(orders.c.total).where(orders.c.user_id == 1)
))

NULL 处理

# 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 pybatis 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

# 模糊匹配:% 任意字符、_ 单字符
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 pybatis.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

# 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

from pybatis 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 pybatis 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

# 基础分页
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 去重

# 单列去重
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

from pybatis 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 与子查询

from pybatis 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

时间范围查询

from datetime import date, datetime, timedelta
from pybatis 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 字段查询

# PG JSONB
from pybatis.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 写法”。

业务问句

写法

找出所有 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(窗口函数)

from pybatis 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 / 锁

from pybatis 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))

性能提示

# 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 基础

from pybatis 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 定义一对一、一对多、多对多关系。

from pybatis.orm import (
    declarative_base, relationship, Mapped, mapped_column,
)
from pybatis import Integer, String, ForeignKey

Base = declarative_base()

一对一(One-to-One)

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)

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 pybatis.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)

# 关联表(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")

自引用关系

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)

from pybatis.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 策略

from pybatis.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 与生命周期

from pybatis.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 pybatis.orm import scoped_session
SessionScoped = scoped_session(SessionLocal)
s = SessionScoped()         # 同线程复用同一 Session
s.add(user)
s.commit()
SessionScoped.remove()      # 结束线程时清理

# 4. 异步 Session
from pybatis.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

# 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 pybatis import select
stmt = select(User).where(User.age > 18)
for u in s.execute(stmt).scalars():
    print(u.name)

# 复杂条件
from pybatis 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 格式

dialect+driver://user:password@host:port/database?param=value

参数说明:

  • dialect:数据库类型(sqlite、mysql、postgresql、oracle、mssql)

  • driver:驱动名称(小写,可选,省略时使用默认驱动)

  • 查询参数:作为连接参数透传给驱动

SQLite

# 内存库(测试用)
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

# 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

# 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 pybatis.dialects.postgresql import ARRAY, JSONB, HSTORE
from pybatis.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 pybatis.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 pybatis.dialects.postgresql import TSVECTOR
stmt = select(t).where(t.c.doc.op("@@")(func.to_tsquery("python & sql")))

Oracle

# 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 pybatis.dialects.oracle import RAW, NUMBER, CLOB, BLOB, ROWID

Microsoft SQL Server

# 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 pybatis.dialects.mssql import ROWVERSION
t = Table("t", m, Column("id", Integer, primary_key=True), Column("version", ROWVERSION))

方言能力矩阵

能力

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)

# declarative —— 声明式 ORM
from pybatis.ext.declarative import declarative_base

# automap —— 从已有数据库自动生成模型
from pybatis.ext.automap import automap_base
Base = automap_base()
Base.prepare(engine, reflect=True)

# associationproxy —— 集合属性代理
from pybatis.ext.associationproxy import association_proxy
class User(Base):
    keywords: Mapped[list[str]] = association_proxy("user_keywords", "value")

# hybrid —— 同时在 Python 层与 SQL 层可用的属性
from pybatis.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 pybatis.ext.mutable import MutableDict, MutableList
class User(Base):
    preferences: Mapped[dict] = mapped_column(MutableDict.as_mutable(JSON))

# orderinglist —— 有序集合
from pybatis.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 pybatis.ext.serializer import PickleType
blobs: Mapped[dict] = mapped_column(PickleType)

# horizontal_shard —— 横向分片
from pybatis.ext.horizontal_shard import ShardedSession
# 通过 create_engine + 路由函数实现分库分表

# baked —— 预编译查询缓存
from pybatis.ext.baked import baked

# indexable —— 支持 ``obj["key"]`` 访问的字典类型
from pybatis.ext.indexable import indexable

# asyncio —— 异步 IO 支持
from pybatis.ext.asyncio import (
    create_async_engine, AsyncSession, async_sessionmaker,
)

扩展模块:雪花 ID(Snowflake)

# 导入
from pybatis.ext.snowflake import (
    snowflake_id,            # 便捷函数
    SnowflakeID,             # Column 类型
    SnowflakeGenerator,      # 完整生成器类
    snowflake_column,        # 工厂函数
    configure_default,       # 配置进程默认生成器
    TWITTER_EPOCH,           # 起始时间戳常量
)

ID 结构

64 位二进制按 BigInteger 存储:

位

含义

位

含义

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

from pybatis.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)

from pybatis import Table, Column, MetaData, String
from pybatis.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 pybatis.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)),
)

工厂函数(更短)

from pybatis.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 用法

from pybatis.orm import Base, Mapped, mapped_column, Session
from pybatis.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)     # 自动生成

进程级默认生成器

from pybatis.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 pybatis.ext.snowflake import snowflake_id
uid = snowflake_id()    # 使用默认生成器

# 运行时替换(不推荐)
set_default_generator(SnowflakeGenerator(worker_id=5, datacenter_id=5))

线程安全

所有 ID 生成操作都通过内部锁保证线程安全;4 线程 × 100 次共 400 个 ID 全部唯一且递增。

import threading
from pybatis.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    # 全部唯一

时钟回拨处理

# 容忍 ≤ 5ms 的回拨(自旋等待追上)
# 超出则抛 ClockMovedBackwardsError
from pybatis.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,而是封装一个字段:

from pybatis import Table, Column, MetaData, String
from pybatis.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)),
)

配置与连接池

# 调试:打印 SQL
engine = create_engine("postgresql+psycopg://...", echo=True)
engine.echo = True  # 运行时切换

# 只打印部分
import logging
logging.basicConfig()
logging.getLogger("pybatis.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()

事件系统

from pybatis 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 pybatis.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

类型与自定义

from pybatis 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 pybatis.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 定义:

from pybatis 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:

pip install alembic
alembic init migrations
# 在 alembic/env.py 中将 target_metadata 指向 Base.metadata

使用示例

以下 4 个示例覆盖最常见的使用场景,全部可独立运行。

完整博客系统(端到端)

一个含用户、文章、评论、标签(多对多)的完整博客系统,从建表到查询一气呵成。

from datetime import datetime
from pybatis import (
    create_engine, MetaData, Table, Column, Integer, String, Text,
    ForeignKey, DateTime, Boolean, select, insert, update, delete,
    func, and_, or_,
)
from pybatis.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。

# main.py
from typing import Generator
from fastapi import Depends, FastAPI, HTTPException
from pybatis import create_engine
from pybatis.orm import Session, sessionmaker
from pybatis.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 pybatis 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 pybatis.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 pybatis 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 集成

# app.py
from flask import Flask, g
from pybatis import create_engine
from pybatis.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 pybatis 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/<int:item_id>", 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 之外的存储。

# 在 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 pybatis import create_engine
from pybatis.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 pybatis 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)

生产部署清单

# 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 pybatis 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 pybatis 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]

测试用例

# 用 pytest + 内存 SQLite 跑 PyBatis 应用测试
# conftest.py
import pytest
from pybatis import create_engine
from pybatis.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 pybatis 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% 提升。

pip install cython>=3.3
pip install -e .    # 自动编译 .pyx

编译产物在 lib/pybatis/engine/_processors_cy.cpython-*.so、_row_cy.so、_result_cy.so、_cache_key_cy.so、_util_cy.so、_collections_cy.so、_immutabledict_cy.so。

# 运行时检查 C 扩展是否启用
import pybatis
print(pybatis.__version__)
import pybatis.engine._processors_cy as _
print("C extension: enabled")  # 没编译会 ImportError

连接池调优

# 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 pybatis.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 条子查询。

# 错:触发 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 pybatis.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 pybatis.orm import joinedload
stmt = select(Post).options(joinedload(Post.author))

# 对 3:subqueryload(1 条子查询)
from pybatis.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 pybatis.orm import contains_eager
stmt = (
    select(User)
    .join(Post)
    .where(Post.published == True)
    .options(contains_eager(User.posts))
)

只取需要的列

# 错:取全部列(包括大字段)
stmt = select(User)                  # 包含 bio(Text)、avatar(LargeBinary)等

# 对:只取需要的列
stmt = select(User.id, User.username, User.email)

# ORM 也支持列级限定
from pybatis.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 个层级的批量能力。

# 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 pybatis 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()

流式读取大数据集

处理百万行结果时不要一次性加载到内存。

# 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 配置

# 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)

事务粒度

# 错:每条 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 层优化

# 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 pybatis 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 pybatis import text
plan = s.execute(text("EXPLAIN ANALYZE SELECT * FROM users WHERE name = :n"),
                 {"n": "alice"}).fetchall()
for row in plan:
    print(row[0])

异步 + 并发

# 1. 异步批量并发
import asyncio
from pybatis.ext.asyncio import create_async_engine, AsyncSession

async def fetch_user(s, uid):
    from pybatis 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://...")

诊断与监控

# 1. echo SQL
engine = create_engine(url, echo=True)         # 全部 SQL
engine.echo = False                              # 运行时关闭

# 2. 部分 SQL 日志
import logging
logging.basicConfig()
logging.getLogger("pybatis.engine").setLevel(logging.INFO)
logging.getLogger("pybatis.pool").setLevel(logging.DEBUG)   # 连接池事件
logging.getLogger("pybatis.orm").setLevel(logging.INFO)     # ORM 事件

# 3. 慢查询捕获(自定义事件)
from pybatis 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"))

常见性能陷阱

陷阱

修法

触发 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 大致能达到:

操作

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 pybatis`` 报 ModuleNotFoundError?

确认已通过 pip install -e . 或 pip install pybatis 安装;或在脚本中手动 sys.path.insert(0, "lib")。

Q: SQLite 文件被占用?

默认 SQLite 仅允许单线程访问同一连接。多线程需 connect_args={"check_same_thread": False} 或使用 StaticPool。

Q: 如何处理 Pymysql / 异步驱动的警告?

设置 pybatis_warn_20 = "false" 或在环境变量中 export PYBATIS_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 日志?

import logging
logging.getLogger("pybatis.engine").setLevel(logging.WARNING)

Q: 异步与同步能否混用?

PyBatis 的同步 Engine 和 AsyncEngine 互不通用,需分别创建。同一进程内可同时存在两种 Session(绑定不同 Engine)。


版本与许可

当前版本:2.1.5

协议:MIT(详见 LICENSE 文件)

Metadata

Release files for pybatis-plus 1.0.0

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for pybatis-plus 1.0.0
File Size Uploaded
pybatis_plus-1.0.0.tar.gz 5.5 MB Details

Built distribution (wheel)

Table of built distributions (wheels) for pybatis-plus 1.0.0
File Interpreter ABI Platform
pybatis_plus-1.0.0-py3-none-any.whl Python 3 none any Details

Total release size: 7.6 MB

Release files / pybatis_plus-1.0.0.tar.gz

Download URL pybatis_plus-1.0.0.tar.gz
Size 5.5 MB
Tags Source
SHA-256 checksum
How to use checksums
e40a53e7144c759f390e9ed9ae22c813b68ae19bf20cb133c37f7b179de555a1
BLAKE2b-256 checksum
How to use checksums
14517ce6958f7ef31de5df8059a89c42298b4cfb0c21ed4be5a55e0f86035c68
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No
Uploaded via twine/7.0.0 CPython/3.13.12

Release files / pybatis_plus-1.0.0-py3-none-any.whl

Download URL pybatis_plus-1.0.0-py3-none-any.whl
Size 2.1 MB
Tags Python 3
SHA-256 checksum
How to use checksums
d1d02a1f34647361fcdc3d9c177ab909a9519a95c3b06fb6003a5f009f7c94fe
BLAKE2b-256 checksum
How to use checksums
19a1bbc4c644f8d791602715c147146ff010d75172025b1d92d6f1a5f51b8aec
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No
Uploaded via twine/7.0.0 CPython/3.13.12

Release history Release notifications | RSS feed

1.0.1

2 release files

This release

1.0.0 This release

2 release files

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page