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

数据类型速查

增（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)

比较运算符

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

逻辑组合 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 写法”。

分组内 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",
    }