Metadata-Version: 2.4
Name: pbatis
Version: 1.1.3
Summary: Python 数据库工具与对象关系映射（ORM）框架
Author-email: pengdeyou <82176754@qq.com>
License-Expression: MIT
Project-URL: Homepage, https://www.pbatis.org
Project-URL: Documentation, https://docs.pbatis.org
Project-URL: Changelog, https://docs.pbatis.org/latest/changelog/index.html
Project-URL: Source Code, https://gitee.com/pbatis/pbatis
Project-URL: Issue Tracker, https://gitee.com/pbatis/pbatis/issues
Project-URL: Discussions, https://gitee.com/pbatis/pbatis/issues
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: Natural Language :: Chinese (Simplified)
Classifier: Natural Language :: English
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: 3.15
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Programming Language :: Python :: Implementation :: PyPy
Classifier: Topic :: Database :: Front-Ends
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.11
Description-Content-Type: text/x-rst
License-File: LICENSE
License-File: AUTHORS
Requires-Dist: typing-extensions>=4.6.0
Provides-Extra: asyncio
Requires-Dist: greenlet>=1; extra == "asyncio"
Provides-Extra: mypy
Requires-Dist: mypy>=2.4; extra == "mypy"
Requires-Dist: types-greenlet>=2; extra == "mypy"
Provides-Extra: mssql
Requires-Dist: pyodbc; extra == "mssql"
Provides-Extra: mssql-pymssql
Requires-Dist: pymssql; extra == "mssql-pymssql"
Provides-Extra: mssql-pyodbc
Requires-Dist: pyodbc; extra == "mssql-pyodbc"
Provides-Extra: mssql-python
Requires-Dist: mssql-python>=1.9.0; extra == "mssql-python"
Provides-Extra: mysql
Requires-Dist: mysqlclient>=1.4.0; extra == "mysql"
Provides-Extra: mysql-connector
Requires-Dist: mysql-connector-python; extra == "mysql-connector"
Provides-Extra: mariadb-connector
Requires-Dist: mariadb!=1.1.10,!=1.1.2,!=1.1.5,>=1.0.1; extra == "mariadb-connector"
Provides-Extra: oracle
Requires-Dist: oracledb>=2.0.1; extra == "oracle"
Provides-Extra: oracle-cxoracle
Requires-Dist: cx_oracle>=8; extra == "oracle-cxoracle"
Provides-Extra: oracle-oracledb
Requires-Dist: oracledb>=2.0.1; extra == "oracle-oracledb"
Provides-Extra: postgresql
Requires-Dist: psycopg!=3.1.15,>=3.0.7; extra == "postgresql"
Provides-Extra: postgresql-pg8000
Requires-Dist: pg8000>=1.29.3; extra == "postgresql-pg8000"
Provides-Extra: postgresql-asyncpg
Requires-Dist: pbatis[asyncio]; extra == "postgresql-asyncpg"
Requires-Dist: asyncpg; extra == "postgresql-asyncpg"
Provides-Extra: postgresql-psycopg2binary
Requires-Dist: psycopg2-binary; extra == "postgresql-psycopg2binary"
Provides-Extra: postgresql-psycopg2cffi
Requires-Dist: psycopg2cffi; extra == "postgresql-psycopg2cffi"
Provides-Extra: postgresql-psycopg
Requires-Dist: psycopg!=3.1.15,>=3.0.7; extra == "postgresql-psycopg"
Provides-Extra: postgresql-psycopgbinary
Requires-Dist: psycopg[binary]!=3.1.15,>=3.0.7; extra == "postgresql-psycopgbinary"
Provides-Extra: pymysql
Requires-Dist: pymysql; extra == "pymysql"
Provides-Extra: cymysql
Requires-Dist: cymysql; extra == "cymysql"
Provides-Extra: aiomysql
Requires-Dist: pbatis[asyncio]; extra == "aiomysql"
Requires-Dist: aiomysql; extra == "aiomysql"
Provides-Extra: aioodbc
Requires-Dist: pbatis[asyncio]; extra == "aioodbc"
Requires-Dist: aioodbc; extra == "aioodbc"
Provides-Extra: asyncmy
Requires-Dist: pbatis[asyncio]; extra == "asyncmy"
Requires-Dist: asyncmy>=0.2.12; extra == "asyncmy"
Provides-Extra: aiosqlite
Requires-Dist: pbatis[asyncio]; extra == "aiosqlite"
Requires-Dist: aiosqlite; extra == "aiosqlite"
Provides-Extra: sqlcipher
Requires-Dist: sqlcipher3_binary; extra == "sqlcipher"
Dynamic: license-file

PyBatis
=======

.. image:: docs/pbatis-logo.png
   :alt: PyBatis ORM Logo
   :align: center
   :width: 280px

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

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

----

安装
----

.. code-block:: bash

    pip install pbatis

按需安装特定方言驱动：

.. code-block:: bash

    # SQLite（标准库自带 pysqlite，无需额外安装）
    pip install pbatis[aiosqlite]        # 异步 SQLite
    pip install pbatis[sqlcipher]        # 加密 SQLite

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

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

    # Oracle
    pip install pbatis[oracle]            # oracledb（推荐）
    pip install pbatis[oracle-cxoracle]   # cx_Oracle

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

    # 异步支持（greenlet）
    pip install pbatis[asyncio]

要求 Python 3.11 及以上。

----

快速开始
--------

.. code-block:: python

    from pbatis import (
        create_engine, MetaData, Table, Column,
        Integer, String, ForeignKey, select, insert, update, delete,
    )

    # 1. 建表（Schema）
    metadata = MetaData()

    users = Table(
        "users", metadata,
        Column("id", Integer, primary_key=True),
        Column("name", String(50), nullable=False),
        Column("email", String(120), unique=True),
    )

    addresses = Table(
        "addresses", metadata,
        Column("id", Integer, primary_key=True),
        Column("user_id", Integer, ForeignKey("users.id")),
        Column("city", String(80)),
    )

    # 2. 创建引擎（Engine = 连接池 + 方言）
    engine = create_engine("sqlite:///:memory:", echo=True)

    # 3. 建表
    metadata.create_all(engine)

    # 4. 插入数据
    with engine.begin() as conn:
        conn.execute(insert(users).values(name="张三", email="a@example.com"))
        conn.execute(insert(addresses).values(user_id=1, city="深圳"))

    # 5. 查询
    with engine.connect() as conn:
        for row in conn.execute(select(users)):
            print(row.id, row.name, row.email)

----

核心概念
--------

Engine —— 数据库引擎
~~~~~~~~~~~~~~~~~~~~~

``Engine`` 是 PyBatis 访问数据库的入口，内部维护一个连接池。

.. code-block:: python

    from pbatis import create_engine

    # SQLite 内存库（测试用）
    engine = create_engine("sqlite:///:memory:")

    # SQLite 文件库
    engine = create_engine("sqlite:///data/app.db")

    # MySQL
    engine = create_engine("mysql+pymysql://user:password@localhost:3306/mydb?charset=utf8mb4")

    # PostgreSQL
    engine = create_engine("postgresql+psycopg://user:password@localhost:5432/mydb")

    # 通用参数
    engine = create_engine(
        "postgresql+psycopg://localhost/mydb",
        echo=True,             # 打印 SQL
        pool_size=5,           # 连接池大小
        max_overflow=10,       # 超出池大小时允许的临时连接
        pool_recycle=3600,     # 连接回收秒数
        pool_pre_ping=True,    # 取出连接前 ping 一下
    )

Connection —— 数据库连接
~~~~~~~~~~~~~~~~~~~~~~~~~

``Connection`` 由 Engine 提供，支持上下文管理器自动释放。

.. code-block:: python

    # 自动提交（事务）
    with engine.begin() as conn:
        conn.execute(insert(users).values(name="李四"))
        conn.execute(insert(addresses).values(user_id=2, city="广州"))
        # 退出 with 块自动 commit；异常自动 rollback

    # 手动事务
    with engine.connect() as conn:
        conn.execute(insert(users).values(name="王五"))
        conn.commit()    # 显式提交
        # 或 conn.rollback()

    # 同一连接上多次执行
    with engine.connect() as conn:
        result = conn.execute(select(users))
        for row in result:
            print(row)

    # 异步连接
    from pbatis.ext.asyncio import create_async_engine
    async with create_async_engine("sqlite+aiosqlite:///:memory:").begin() as conn:
        await conn.execute(insert(users).values(name="赵六"))

Schema —— 表与列定义
~~~~~~~~~~~~~~~~~~~~

使用 ``Table`` / ``Column`` 直接定义；或在 ORM 模式下用 ``declarative_base()`` 声明。

.. code-block:: python

    from pbatis import (
        MetaData, Table, Column, Integer, String,
        ForeignKey, DateTime, Numeric, Boolean, func,
    )

    metadata = MetaData()

    # 基础类型
    products = Table(
        "products", metadata,
        Column("id", Integer, primary_key=True, autoincrement=True),
        Column("sku", String(32), unique=True, nullable=False),
        Column("name", String(200), nullable=False),
        Column("price", Numeric(10, 2), default=0),
        Column("active", Boolean, default=True),
        Column("created_at", DateTime, server_default=func.now()),
        Column("updated_at", DateTime, onupdate=func.now()),
    )

    # 主键约束、唯一约束、检查约束
    from pbatis import UniqueConstraint, CheckConstraint, Index

    accounts = Table(
        "accounts", metadata,
        Column("id", Integer, primary_key=True),
        Column("email", String(120)),
        Column("age", Integer),
        UniqueConstraint("email", name="uq_accounts_email"),
        CheckConstraint("age >= 0", name="ck_accounts_age"),
        Index("ix_accounts_email", "email"),
    )

    # 复合主键
    order_items = Table(
        "order_items", metadata,
        Column("order_id", Integer, primary_key=True),
        Column("line_no", Integer, primary_key=True),
        Column("sku", String(32)),
    )

    # 跨表外键
    orders = Table(
        "orders", metadata,
        Column("id", Integer, primary_key=True),
        Column("user_id", Integer, ForeignKey("users.id", ondelete="CASCADE")),
        Column("total", Numeric(10, 2)),
    )

数据类型速查
~~~~~~~~~~~~

.. list-table::
   :header-rows: 1
   :widths: 20 30 50

   * - 类别
     - 类型
     - 适用场景
   * - 整数
     - ``Integer``、``BigInteger``、``SmallInteger``
     - 主键、外键、计数
   * - 字符串
     - ``String``、``Text``、``Unicode``、``UnicodeText``
     - 短/长文本，Unicode 显式指定字符集
   * - 浮点
     - ``Float``、``Numeric``、``Double``
     - 金额用 ``Numeric`` 避免二进制浮点误差
   * - 日期时间
     - ``Date``、``Time``、``DateTime``、``Interval``
     - 时间戳、时段
   * - 布尔
     - ``Boolean``
     - 真假标志位
   * - JSON
     - ``JSON``
     - 文档型字段（方言支持差异见下）
   * - 二进制
     - ``LargeBinary``
     - 字节流、BLOB
   * - 枚举
     - ``Enum``、``PickleType``
     - 有限集合
   * - 数组（PG）
     - ``ARRAY(Integer)``
     - PostgreSQL 原生数组
   * - 自定义
     - ``TypeDecorator``、``UserDefinedType``
     - 包装或自定义类型映射

----

SQL 表达式语言（Core）
----------------------

增（INSERT）
~~~~~~~~~~~~

.. code-block:: python

    from pbatis import insert

    # 单行插入
    stmt = insert(users).values(name="张三", email="a@example.com")
    with engine.begin() as conn:
        result = conn.execute(stmt)
        print(result.inserted_primary_key)  # 自动生成的主键

    # 多行批量插入（性能更高）
    with engine.begin() as conn:
        conn.execute(insert(users), [
            {"name": "李四", "email": "b@example.com"},
            {"name": "王五", "email": "c@example.com"},
            {"name": "赵六", "email": "d@example.com"},
        ])

    # INSERT ... SELECT（从另一张表查出来再插入）
    with engine.begin() as conn:
        conn.execute(
            insert(archived_users).select().where(users.c.active == False)
        )

    # INSERT ... ON CONFLICT（PG）/ ON DUPLICATE KEY（MySQL）/ MERGE（Oracle）
    from pbatis.dialects.postgresql import insert as pg_insert

    stmt = pg_insert(users).values(id=1, name="冲突", email="x@x.com")
    upsert = stmt.on_conflict_do_update(
        index_elements=[users.c.id],
        set_={"name": stmt.excluded.name},
    )
    with engine.begin() as conn:
        conn.execute(upsert)

    # MySQL 等价写法
    from pbatis.dialects.mysql import insert as mysql_insert
    stmt = mysql_insert(users).values(id=1, name="冲突", email="x@x.com")
    upsert = stmt.on_duplicate_key_update(name=stmt.inserted.name)

    # 返回值（RETURNING）
    from pbatis import insert
    stmt = insert(users).values(name="回写").returning(users.c.id, users.c.name)
    with engine.connect() as conn:
        for row in conn.execute(stmt):
            print(row)

删（DELETE）
~~~~~~~~~~~~

.. code-block:: python

    from pbatis import delete

    # 按条件删除
    with engine.begin() as conn:
        conn.execute(delete(users).where(users.c.id == 5))

    # 多条件
    with engine.begin() as conn:
        conn.execute(
            delete(users)
            .where(users.c.active == False)
            .where(users.c.created_at < "2020-01-01")
        )

    # RETURNING（PostgreSQL / SQLite / Oracle 21c+）
    stmt = delete(users).where(users.c.id == 5).returning(users.c.id)
    with engine.connect() as conn:
        for row in conn.execute(stmt):
            print("deleted", row.id)

改（UPDATE）
~~~~~~~~~~~~

.. code-block:: python

    from pbatis import update

    # 单字段
    with engine.begin() as conn:
        conn.execute(
            update(users)
            .where(users.c.id == 1)
            .values(name="张三丰")
        )

    # 多字段
    with engine.begin() as conn:
        conn.execute(
            update(users)
            .where(users.c.email == "old@example.com")
            .values(name="新名", active=False, updated_at=func.now())
        )

    # 批量按表达式更新
    from pbatis import case
    with engine.begin() as conn:
        conn.execute(
            update(products)
            .where(products.c.category == "books")
            .values(price=case(
                (products.c.price < 50, products.c.price * 1.1),
                (products.c.price < 100, products.c.price * 1.05),
                else_=products.c.price,
            ))
        )

    # RETURNING
    stmt = (
        update(users)
        .where(users.c.id == 1)
        .values(name="新")
        .returning(users.c.id, users.c.name)
    )
    with engine.connect() as conn:
        for row in conn.execute(stmt):
            print(row)

查（SELECT）
~~~~~~~~~~~~

.. code-block:: python

    from pbatis import select, and_, or_, not_

    # 基础查询
    stmt = select(users.c.id, users.c.name).where(users.c.active == True)
    with engine.connect() as conn:
        for row in conn.execute(stmt):
            print(row.id, row.name)

    # 排序
    stmt = select(users).order_by(users.c.created_at.desc(), users.c.id.asc())

    # LIMIT / OFFSET
    stmt = select(users).limit(10).offset(20)

    # DISTINCT
    stmt = select(users.c.city).distinct()

    # 聚合
    from pbatis import func
    stmt = select(
        users.c.city,
        func.count(users.c.id).label("cnt"),
        func.avg(users.c.age).label("avg_age"),
    ).group_by(users.c.city).having(func.count(users.c.id) > 5)

    # 子查询
    subq = select(func.max(orders.c.total)).scalar_subquery()
    stmt = select(orders).where(orders.c.total == subq)

    # UNION
    from pbatis import union, union_all
    stmt = union(
        select(users.c.email).where(users.c.active == True),
        select(addresses.c.email).where(addresses.c.primary == True),
    )

    # CTE（公共表表达式）
    cte = (
        select(orders.c.user_id, func.sum(orders.c.total).label("total"))
        .group_by(orders.c.user_id)
        .cte("user_totals")
    )
    stmt = select(users.c.name, cte.c.total).join(cte, cte.c.user_id == users.c.id)

----

查询条件与过滤
--------------

WHERE 子句里能写什么——一次性讲清。

比较运算符
~~~~~~~~~~

.. code-block:: python

    from pbatis import select

    # 等于 =
    select(users).where(users.c.name == "张三")
    # 渲染：WHERE name = '张三'

    # 不等于（两种写法等价）
    select(users).where(users.c.name != "张三")
    select(users).where(users.c.name.op("!=")("张三"))

    # 大于 / 小于 / 大于等于 / 小于等于
    select(products).where(products.c.price > 100)            # 大于
    select(products).where(products.c.price < 100)            # 小于
    select(products).where(products.c.price >= 100)           # 大于等于
    select(products).where(products.c.price <= 100)           # 小于等于

    # 区间 BETWEEN（闭区间，含两端）
    from pbatis import between
    select(products).where(products.c.price.between(50, 200))
    # 渲染：WHERE price BETWEEN 50 AND 200

    # 排除区间（NOT BETWEEN）
    select(products).where(~products.c.price.between(50, 200))

    # 同值多个字段（row-value 比较）
    select(orders).where(
        tuple_(orders.c.region, orders.c.priority) == tuple_("CN", 1)
    )

运算符速查
^^^^^^^^^^

.. list-table::
   :header-rows: 1
   :widths: 20 30 50

   * - 运算符
     - Python 写法
     - 生成的 SQL
   * - 等于
     - ``col == "x"``
     - ``col = 'x'``
   * - 不等于
     - ``col != "x"``
     - ``col <> 'x'``
   * - 大于 / 小于
     - ``col > 10``、``col < 10``
     - ``col > 10``、``col < 10``
   * - 大于等于 / 小于等于
     - ``col >= 10``、``col <= 10``
     - ``col >= 10``、``col <= 10``
   * - BETWEEN
     - ``col.between(a, b)``
     - ``col BETWEEN a AND b``
   * - IN
     - ``col.in_([...])``
     - ``col IN (...)``
   * - NOT IN
     - ``~col.in_([...])``
     - ``col NOT IN (...)``
   * - IS NULL
     - ``col.is_(None)``
     - ``col IS NULL``
   * - IS NOT NULL
     - ``col.isnot(None)``
     - ``col IS NOT NULL``
   * - LIKE
     - ``col.like("%foo%")``
     - ``col LIKE '%foo%'``
   * - ILIKE（PG）
     - ``col.ilike("%foo%")``
     - ``col ILIKE '%foo%'``
   * - 真值
     - ``col.is_(True)``
     - ``col IS true``

逻辑组合 AND / OR / NOT
~~~~~~~~~~~~~~~~~~~~~~~~

.. code-block:: python

    from pbatis import and_, or_, not_

    # 1. 隐式 AND（链式 .where()）
    select(users).where(
        users.c.age > 18,
        users.c.city == "深圳",
        users.c.active == True,
    )
    # 等价于：WHERE age > 18 AND city = '深圳' AND active = true

    # 2. 显式 and_(...)
    select(users).where(and_(
        users.c.age > 18,
        users.c.city == "深圳",
    ))

    # 3. OR
    select(users).where(or_(
        users.c.city == "深圳",
        users.c.city == "广州",
    ))
    # 渲染：WHERE city = '深圳' OR city = '广州'

    # 4. 混合
    select(users).where(and_(
        users.c.age > 18,
        or_(
            users.c.city == "深圳",
            users.c.city == "广州",
        ),
        users.c.active == True,
    ))
    # 渲染：WHERE age > 18 AND (city = '深圳' OR city = '广州') AND active = true

    # 5. NOT（取反，前缀 ~）
    select(users).where(~users.c.active)               # 活跃 = false
    select(users).where(not_(users.c.age > 18))        # 18 岁以下

    # 6. 用布尔对象直接取反
    from pbatis import true, false, null
    select(users).where(users.c.active == false)
    select(users).where(users.c.active.is_not(True))

    # 7. any_/all_ 配合子查询（PG、MySQL 8+）
    from pbatis import any_, all_
    select(products).where(products.c.price < any_(
        select(orders.c.total).where(orders.c.user_id == 1)
    ))

NULL 处理
~~~~~~~~~

.. code-block:: python

    # IS NULL
    select(users).where(users.c.email.is_(None))
    # 渲染：WHERE email IS NULL

    # IS NOT NULL
    select(users).where(users.c.email.isnot(None))
    # 渲染：WHERE email IS NOT NULL

    # 不要用 == None（不会生成 IS NULL）
    # select(users).where(users.c.email == None)  # 错：会变成 email = NULL（永远不匹配）

    # COALESCE（取第一个非 NULL）
    from pbatis import func
    select(func.coalesce(users.c.nickname, users.c.name, "匿名"))

    # NULLIF（两值相等返回 NULL）
    select(func.nullif(users.c.status, "deleted"))

    # 字段是否为 NULL 的布尔返回（PG / MySQL）
    select(users).where(users.c.deleted_at.is_(None))        # 未删除
    select(users).where(users.c.deleted_at.isnot(None))      # 已删除

字符串匹配 LIKE / ILIKE / REGEX
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

.. code-block:: python

    # 模糊匹配：% 任意字符、_ 单字符
    select(users).where(users.c.name.like("张%"))         # 张某、张三丰
    select(users).where(users.c.name.like("%三%"))        # 包含"三"
    select(users).where(users.c.name.like("张_"))         # 张三（两字）

    # 不区分大小写（PG 用 ilike，MySQL/SQLite 用 lower()）
    select(users).where(users.c.name.ilike("%ZHANG%"))         # PG
    select(users).where(users.c.name.lower().like("%zhang%")) # 跨方言

    # 取反
    select(users).where(~users.c.name.like("张%"))           # NOT LIKE

    # 转义 LIKE 通配符
    select(users).where(
        users.c.filename.like("%10\%%", escape="\\")
    )

    # 正则（PG、Oracle、MySQL 8+）
    select(users).where(users.c.email.op("~")("^[a-z]+@"))    # PG
    select(users).where(users.c.name.op("REGEXP")("^张"))    # MySQL

    # 全文搜索（PG）
    from pbatis.dialects.postgresql import TSVECTOR
    stmt = (
        select(articles)
        .where(articles.c.searchable.op("@@")(func.to_tsquery("python & sql")))
    )

    # 全文搜索（MySQL）
    stmt = select(articles).where(articles.c.body.op("MATCH")("python sql"))

集合操作 IN / NOT IN / ANY / ALL
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

.. code-block:: python

    # IN：值在列表内
    select(users).where(users.c.city.in_(["深圳", "广州", "上海"]))
    # 渲染：WHERE city IN ('深圳', '广州', '上海')

    # NOT IN
    select(users).where(~users.c.city.in_(["北京", "上海"]))

    # IN 子查询
    select(users).where(users.c.id.in_(
        select(orders.c.user_id).where(orders.c.total > 1000)
    ))

    # 元组 IN（PG、MySQL）
    select(orders).where(
        tuple_(orders.c.region, orders.c.status).in_([
            ("CN", "paid"),
            ("HK", "shipped"),
        ])
    )

    # 任意一个满足（= ANY IN）
    select(products).where(products.c.tag_id == any_([1, 2, 3]))
    # 渲染：WHERE tag_id = ANY(ARRAY[1, 2, 3])

    # 全部满足（ALL）
    select(products).where(products.c.price > all_([100, 200, 300]))

排序 ORDER BY
~~~~~~~~~~~~~

.. code-block:: python

    from pbatis import asc, desc, nulls_first, nulls_last

    # 单列升序 / 降序
    select(users).order_by(users.c.name.asc())
    select(users).order_by(users.c.name.desc())
    # 简写
    select(users).order_by(users.c.name)         # 默认 ASC
    select(users).order_by(users.c.created_at.desc())

    # 多列排序（按列表顺序生效）
    select(users).order_by(
        users.c.active.desc(),          # 活跃用户排前
        users.c.created_at.desc(),      # 同状态按时间倒序
        users.c.id.asc(),               # 同时间按 id 升序
    )

    # NULL 位置（PG、Oracle、MSSQL；MySQL 8+）
    select(users).order_by(users.c.last_login.desc().nulls_last())
    # last_login IS NULL 排末尾

    select(users).order_by(users.c.last_login.desc().nulls_first())
    # last_login IS NULL 排前面

    # 表达式排序
    select(products).order_by(
        func.length(products.c.name).asc(),  # 按名称长度排
    )
    select(users).order_by(
        func.random(),                  # 随机（SQLite/PG）
    )
    select(users).order_by(
        func.rand(),                    # 随机（MySQL）
    )

    # CASE 排序（自定义优先级）
    from pbatis import case
    priority = case(
        (users.c.role == "admin", 1),
        (users.c.role == "vip", 2),
        else_=3,
    )
    select(users).order_by(priority.asc(), users.c.id.asc())

分页 LIMIT / OFFSET
~~~~~~~~~~~~~~~~~~~~

.. code-block:: python

    # 基础分页
    stmt = select(users).order_by(users.c.id).limit(10).offset(20)
    # 跳过前 20 条，取 10 条

    # 仅取前 N 条
    select(users).order_by(users.c.id).limit(5)

    # 切片（等价 limit + offset）
    select(users).order_by(users.c.id).slice(20, 30)   # skip 20, take 10

    # 大数据集分页——用 WHERE 替代 OFFSET（性能更好）
    # 慢：OFFSET 越大越慢
    # select(users).where(...).offset(1000000)
    # 快：键集分页（seek pagination）
    last_id = 1000000
    select(users).where(users.c.id > last_id).order_by(users.c.id).limit(10)

    # 流式读取（处理超大数据集）
    with engine.connect() as conn:
        result = conn.execution_options(stream_results=True).execute(
            select(events).order_by(events.c.id)
        )
        for row in result:
            process(row)

DISTINCT 去重
~~~~~~~~~~~~~

.. code-block:: python

    # 单列去重
    select(users.c.city).distinct()
    # 渲染：SELECT DISTINCT city

    # 多列联合去重
    select(users.c.city, users.c.gender).distinct()
    # city + gender 都相同才算重复

    # 配合聚合
    select(func.count(func.distinct(users.c.city))).select_from(users)
    # 不同城市数量

聚合与 HAVING
~~~~~~~~~~~~~~

.. code-block:: python

    from pbatis import func

    # 基础聚合
    select(func.count(users.c.id))                          # 总行数
    select(func.count(func.distinct(users.c.city)))         # 不同城市数
    select(func.sum(orders.c.total))                        # 总额
    select(func.avg(orders.c.total))                        # 平均
    select(func.min(orders.c.total), func.max(orders.c.total))  # 最小/最大

    # 分组（GROUP BY）
    stmt = (
        select(
            users.c.city,
            func.count(users.c.id).label("cnt"),
        )
        .group_by(users.c.city)
    )
    # SELECT city, COUNT(id) AS cnt FROM users GROUP BY city

    # 多列分组
    stmt = (
        select(
            users.c.city,
            users.c.gender,
            func.count(users.c.id),
        )
        .group_by(users.c.city, users.c.gender)
    )

    # HAVING（分组后过滤）
    stmt = (
        select(users.c.city, func.count(users.c.id).label("cnt"))
        .group_by(users.c.city)
        .having(func.count(users.c.id) > 100)
    )
    # HAVING COUNT(id) > 100

    # WHERE + GROUP + HAVING
    stmt = (
        select(users.c.city, func.avg(users.c.age).label("avg_age"))
        .where(users.c.active == True)
        .group_by(users.c.city)
        .having(func.avg(users.c.age) > 30)
        .order_by(func.avg(users.c.age).desc())
    )

EXISTS 与子查询
~~~~~~~~~~~~~~~

.. code-block:: python

    from pbatis import select, exists, not_

    # EXISTS：相关子查询有结果即为真
    stmt = select(users).where(
        exists().where(orders.c.user_id == users.c.id)
    )
    # 找出有订单的用户

    # NOT EXISTS
    stmt = select(users).where(
        ~exists().where(orders.c.user_id == users.c.id)
    )
    # 找出没有任何订单的用户

    # 标量子查询（返回单值）
    subq = select(func.max(orders.c.total)).scalar_subquery()
    stmt = select(orders).where(orders.c.total == subq)
    # 找出金额等于最大金额的订单

    # 子查询作为列
    order_count_subq = (
        select(func.count(orders.c.id))
        .where(orders.c.user_id == users.c.id)
        .scalar_subquery()
    )
    stmt = select(users.c.name, order_count_subq.label("order_cnt"))
    # SELECT name, (SELECT COUNT(*) FROM orders WHERE user_id = users.id) AS order_cnt

时间范围查询
~~~~~~~~~~~~

.. code-block:: python

    from datetime import date, datetime, timedelta
    from pbatis import func, between

    # 具体日期
    stmt = select(orders).where(orders.c.created_at >= datetime(2024, 1, 1))

    # 当天
    today = date.today()
    stmt = select(orders).where(func.date(orders.c.created_at) == today)

    # 本周
    stmt = select(orders).where(
        orders.c.created_at >= func.date_sub(func.now(), func.interval(7, "day"))
    )  # MySQL
    stmt = select(orders).where(
        orders.c.created_at >= func.now() - timedelta(days=7)
    )  # 跨方言

    # BETWEEN 时间范围
    start = datetime(2024, 1, 1)
    end = datetime(2024, 1, 31, 23, 59, 59)
    stmt = select(orders).where(orders.c.created_at.between(start, end))

    # 日期函数
    stmt = select(orders).where(func.year(orders.c.created_at) == 2024)
    stmt = select(orders).where(func.month(orders.c.created_at) == 6)
    stmt = select(orders).where(func.date(orders.c.created_at) == "2024-06-01")

    # PG：date_trunc 按月分桶
    stmt = (
        select(
            func.date_trunc("month", orders.c.created_at).label("month"),
            func.sum(orders.c.total),
        )
        .group_by("month")
    )

JSON 字段查询
~~~~~~~~~~~~~~

.. code-block:: python

    # PG JSONB
    from pbatis.dialects.postgresql import JSONB
    users = Table("users", m, Column("profile", JSONB))

    # 顶层 key
    stmt = select(users).where(users.c.profile["city"].astext == "深圳")

    # 嵌套路径
    stmt = select(users).where(
        users.c.profile["address"]["zip"].astext == "518000"
    )

    # 包含（PG JSONB）
    stmt = select(users).where(users.c.profile.contains({"vip": True}))

    # 包含任一 key
    stmt = select(users).where(
        users.c.profile.has_key("preferences")
    )

    # 数组长度（PG JSONB）
    stmt = select(users).where(
        func.jsonb_array_length(users.c.profile["tags"]) > 3
    )

    # MySQL JSON
    stmt = select(users).where(
        users.c.profile["$.city"].astext == "深圳"
    )  # 注意：MySQL 路径以 $. 开头

    # SQLite JSON
    stmt = select(users).where(
        func.json_extract(users.c.profile, "$.city") == "深圳"
    )

查询案例（业务场景 → SQL）
~~~~~~~~~~~~~~~~~~~~~~~~~~

30 个"问业务问题"→"对应 PyBatis 写法"。

.. list-table::
   :header-rows: 1
   :widths: 35 65

   * - 业务问句
     - 写法
   * - 找出所有 30 岁以上的深圳用户
     - ``select(users).where(users.c.age > 30, users.c.city == "深圳")``
   * - 名字里包含"张"字的用户
     - ``select(users).where(users.c.name.like("%张%"))``
   * - 没有邮箱的用户
     - ``select(users).where(users.c.email.is_(None))``
   * - 邮箱不为空的活跃用户
     - ``select(users).where(users.c.email.isnot(None), users.c.active == True)``
   * - 30-50 岁之间的用户
     - ``select(users).where(users.c.age.between(30, 50))``
   * - 订单金额 100-500 之间且已支付
     - ``select(orders).where(orders.c.total.between(100, 500), orders.c.status == "paid")``
   * - 查询在深圳或广州的用户
     - ``select(users).where(users.c.city.in_(["深圳", "广州"]))``
   * - 排除黑名单用户
     - ``select(users).where(~users.c.id.in_(blacklist_ids))``
   * - 有订单记录的用户
     - ``select(users).where(exists().where(orders.c.user_id == users.c.id))``
   * - 从未下单的用户
     - ``select(users).where(~exists().where(orders.c.user_id == users.c.id))``
   * - 订单金额最高的 10 单
     - ``select(orders).order_by(orders.c.total.desc()).limit(10)``
   * - 第 21-40 条用户（按 id 升序）
     - ``select(users).order_by(users.c.id).offset(20).limit(20)``
   * - 一个月内注册的新用户
     - ``select(users).where(users.c.created_at >= func.now() - timedelta(days=30))``
   * - 不同城市数量
     - ``select(func.count(func.distinct(users.c.city)))``
   * - 每个城市的用户数
     - ``select(users.c.city, func.count(users.c.id)).group_by(users.c.city)``
   * - 用户数大于 1000 的城市
     - ``.group_by(users.c.city).having(func.count(users.c.id) > 1000)``
   * - 价格最高的前 3 件商品
     - ``select(products).order_by(products.c.price.desc()).limit(3)``
   * - 按价格区间分组（0-100、100-500、500+）
     - 用 ``case()`` 配合 ``func.sum()`` 实现
   * - 名字长度大于 5 的用户
     - ``select(users).where(func.length(users.c.name) > 5)``
   * - 邮箱域名是 @example.com
     - ``select(users).where(users.c.email.like("%@example.com"))``
   * - 库存为零的商品
     - ``select(products).where(products.c.stock == 0)``
   * - 库存不为零且上架
     - ``select(products).where(products.c.stock > 0, products.c.active == True)``
   * - 模糊搜索（不区分大小写）
     - ``select(users).where(users.c.name.ilike("%query%"))`` (PG) 或 ``.lower().like()``
   * - 找出金额大于平均值的订单
     - ``select(orders).where(orders.c.total > select(func.avg(orders.c.total)).scalar_subquery())``
   * - 每年订单总额
     - ``select(func.extract("year", orders.c.created_at), func.sum(orders.c.total)).group_by(1)``
   * - 找出 7 天内未登录的用户
     - ``select(users).where(or_(users.c.last_login.is_(None), users.c.last_login < func.now() - timedelta(days=7)))``
   * - 同一手机号只返回最新一条
     - 用窗口函数 ``func.row_number().over(partition_by=..., order_by=...)``
   * - 查询 JSON 字段 city = 深圳
     - 见上文"JSON 字段查询"
   * - 关键字全文搜索
     - 见上文"字符串匹配"末尾
   * - 分组后取每组前 N 条
     - 用 CTE + 窗口函数 ``row_number()``

分组内 Top-N（窗口函数）
~~~~~~~~~~~~~~~~~~~~~~~~~

.. code-block:: python

    from pbatis import over, func

    # ROW_NUMBER：每组从 1 开始编号
    rn = func.row_number().over(
        partition_by=orders.c.user_id,        # 分组字段
        order_by=orders.c.total.desc(),      # 组内排序
    ).label("rn")

    # 找出每个用户金额最高的 1 单
    subq = select(orders.c.id, orders.c.user_id, orders.c.total, rn).subquery()
    stmt = select(subq).where(subq.c.rn == 1)

    # RANK：并列同名次会跳号
    rk = func.rank().over(
        partition_by=orders.c.user_id,
        order_by=orders.c.total.desc(),
    )
    stmt = select(orders.c.user_id, orders.c.total, rk.label("rk")).order_by("rk")

    # DENSE_RANK：并列不跳号
    drk = func.dense_rank().over(
        partition_by=orders.c.user_id,
        order_by=orders.c.total.desc(),
    )

    # 移动平均
    ma = func.avg(orders.c.total).over(
        partition_by=orders.c.user_id,
        order_by=orders.c.created_at,
        rows=(None, 6),   # 7 天窗口（含当前行）
    )

    # 累计求和
    cumsum = func.sum(orders.c.total).over(
        partition_by=orders.c.user_id,
        order_by=orders.c.created_at,
    )

    # 配合 WITH 子句取每组前 3
    cte = (
        select(
            orders.c.id,
            orders.c.user_id,
            orders.c.total,
            func.row_number().over(
                partition_by=orders.c.user_id,
                order_by=orders.c.total.desc(),
            ).label("rn"),
        )
        .cte("ranked")
    )
    stmt = select(cte).where(cte.c.rn <= 3).order_by(cte.c.user_id, cte.c.rn)

FOR UPDATE / 锁
~~~~~~~~~~~~~~~

.. code-block:: python

    from pbatis import select

    # 行级排他锁（PG、MySQL InnoDB、Oracle）
    stmt = select(accounts).where(accounts.c.id == 1).with_for_update()
    # 渲染：SELECT ... FOR UPDATE

    # 跳过已被锁的行
    stmt = select(accounts).where(accounts.c.id == 1).with_for_update(skip_locked=True)
    # 渲染：SELECT ... FOR UPDATE SKIP LOCKED

    # 共享锁
    stmt = select(accounts).where(accounts.c.id == 1).with_for_update(read=True)
    # 渲染：SELECT ... FOR SHARE（PG）/ LOCK IN SHARE MODE（MySQL）

    # 在事务中使用
    with engine.begin() as conn:
        acct = conn.execute(select(accounts).where(accounts.c.id == 1).with_for_update()).first()
        acct.balance -= 100
        conn.execute(update(accounts).where(accounts.c.id == 1).values(balance=acct.balance))

性能提示
~~~~~~~~

.. code-block:: python

    # 1. 优先用 = 而不是 LIKE '%xx%'（前缀 % 用不到索引）
    select(users).where(users.c.name.like("张%"))        # OK
    select(users).where(users.c.name.like("%张"))        # 全表扫描

    # 2. 优先用 IN 而不是 OR
    select(users).where(users.c.id.in_([1, 2, 3]))        # OK
    select(users).where(or_(users.c.id == 1, users.c.id == 2))  # 不如 IN

    # 3. 大量分页用键集分页而不是 OFFSET
    # 慢：select(users).offset(1000000)
    # 快：select(users).where(users.c.id > last_seen_id).limit(10)

    # 4. 配合索引字段排序
    # ORDER BY 用索引列（如主键），避免 filesort
    select(users).order_by(users.c.id)              # 走主键
    select(users).order_by(users.c.email)           # 走索引
    select(users).order_by(func.rand())             # 全表扫描

    # 5. 不要 SELECT *，只取需要的列
    select(users.c.id, users.c.name)                # OK
    select(users)                                    # 多取字段

    # 6. 大结果集用流式
    conn.execution_options(stream_results=True).execute(stmt)

----

联表查询
--------

JOIN 基础
~~~~~~~~~

.. code-block:: python

    from pbatis import select, and_

    # INNER JOIN
    stmt = (
        select(users.c.name, addresses.c.city)
        .select_from(users)
        .join(addresses, addresses.c.user_id == users.c.id)
    )

    # LEFT OUTER JOIN
    stmt = (
        select(users.c.name, addresses.c.city)
        .select_from(users)
        .join(addresses, addresses.c.user_id == users.c.id, isouter=True)
    )

    # FULL OUTER JOIN（PG / Oracle）
    stmt = select(users, addresses).join(addresses, ..., full=True)

    # 多表链式 JOIN
    stmt = (
        select(orders)
        .join(users, users.c.id == orders.c.user_id)
        .join(addresses, addresses.c.user_id == users.c.id)
    )

    # ON 子句 + WHERE
    stmt = (
        select(users)
        .join(addresses, addresses.c.user_id == users.c.id)
        .where(addresses.c.city == "深圳")
    )

    # 自连接
    employees = Table("employees", metadata,
        Column("id", Integer, primary_key=True),
        Column("name", String(50)),
        Column("manager_id", Integer, ForeignKey("employees.id")),
    )
    managers = employees.alias("managers")
    stmt = (
        select(employees.c.name.label("emp"), managers.c.name.label("mgr"))
        .join(managers, managers.c.id == employees.c.manager_id)
    )

    # 索引/表名访问
    select(users).where(users.c["my-column"] == 1)   # 列名含连字符时

ORM 关系映射
~~~~~~~~~~~~

PyBatis 提供声明式 ORM，使用 ``declarative_base()`` 配合 ``relationship`` 定义一对一、一对多、多对多关系。

.. code-block:: python

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

    Base = declarative_base()

一对一（One-to-One）
~~~~~~~~~~~~~~~~~~~~

.. code-block:: python

    class User(Base):
        __tablename__ = "users"

        id: Mapped[int] = mapped_column(Integer, primary_key=True)
        name: Mapped[str] = mapped_column(String(50))

        # 一对一：通过 uselist=False
        profile: Mapped["Profile"] = relationship(
            "Profile",
            back_populates="user",
            uselist=False,        # 关键：返回单对象而不是列表
            cascade="all, delete-orphan",
        )

    class Profile(Base):
        __tablename__ = "profiles"

        id: Mapped[int] = mapped_column(Integer, primary_key=True)
        user_id: Mapped[int] = mapped_column(ForeignKey("users.id"), unique=True)
        bio: Mapped[str] = mapped_column(String(500))

        user: Mapped["User"] = relationship("User", back_populates="profile")

    # 增
    with Session(engine) as s:
        u = User(name="张三", profile=Profile(bio="工程师"))
        s.add(u)
        s.commit()

    # 查
    with Session(engine) as s:
        u = s.query(User).filter_by(name="张三").one()
        print(u.profile.bio)   # 直接访问，无需 join

    # 删（cascade 自动清理）
    s.delete(u)
    s.commit()

一对多（One-to-Many）
~~~~~~~~~~~~~~~~~~~~~

.. code-block:: python

    class Author(Base):
        __tablename__ = "authors"

        id: Mapped[int] = mapped_column(Integer, primary_key=True)
        name: Mapped[str] = mapped_column(String(80))

        # 一对多
        books: Mapped[list["Book"]] = relationship(
            "Book",
            back_populates="author",
            cascade="all, delete-orphan",
            order_by="Book.id",
        )

    class Book(Base):
        __tablename__ = "books"

        id: Mapped[int] = mapped_column(Integer, primary_key=True)
        title: Mapped[str] = mapped_column(String(200))
        author_id: Mapped[int] = mapped_column(ForeignKey("authors.id"))

        author: Mapped["Author"] = relationship("Author", back_populates="books")

    # 增
    with Session(engine) as s:
        a = Author(name="莫言", books=[
            Book(title="丰乳肥臀"),
            Book(title="蛙"),
        ])
        s.add(a)
        s.commit()

    # 查（懒加载）
    with Session(engine) as s:
        a = s.query(Author).filter_by(name="莫言").one()
        for b in a.books:
            print(b.title)

    # 预加载（避免 N+1）
    from pbatis.orm import selectinload, joinedload
    stmt = select(Author).options(selectinload(Author.books))
    for a in s.execute(stmt).scalars():
        for b in a.books:
            print(b.title)

    # 反向查
    stmt = select(Book).join(Author).where(Author.name == "莫言")

    # 过滤
    stmt = (
        select(Author)
        .where(Author.books.any(Book.title.like("%丰%")))
    )

多对多（Many-to-Many）
~~~~~~~~~~~~~~~~~~~~~~

.. code-block:: python

    # 关联表（association table）
    student_course = Table(
        "student_course", Base.metadata,
        Column("student_id", Integer, ForeignKey("students.id"), primary_key=True),
        Column("course_id", Integer, ForeignKey("courses.id"), primary_key=True),
        Column("enrolled_at", DateTime, server_default=func.now()),
    )

    class Student(Base):
        __tablename__ = "students"

        id: Mapped[int] = mapped_column(Integer, primary_key=True)
        name: Mapped[str] = mapped_column(String(80))

        courses: Mapped[list["Course"]] = relationship(
            "Course",
            secondary=student_course,    # 关键：指定关联表
            back_populates="students",
        )

    class Course(Base):
        __tablename__ = "courses"

        id: Mapped[int] = mapped_column(Integer, primary_key=True)
        title: Mapped[str] = mapped_column(String(200))

        students: Mapped[list["Student"]] = relationship(
            "Student",
            secondary=student_course,
            back_populates="courses",
        )

    # 增
    with Session(engine) as s:
        math = Course(title="高等数学")
        eng = Course(title="英语")
        s.add(Student(name="小明", courses=[math, eng]))
        s.commit()

    # 增关联（追加）
    with Session(engine) as s:
        stu = s.query(Student).filter_by(name="小明").one()
        physics = s.query(Course).filter_by(title="物理").one()
        stu.courses.append(physics)
        s.commit()

    # 删关联
    stu.courses.remove(math)
    s.commit()

    # 关联表上额外的字段（enrolled_at）—— 使用 Association Object
    class StudentCourse(Base):
        __tablename__ = "student_course"
        student_id: Mapped[int] = mapped_column(ForeignKey("students.id"), primary_key=True)
        course_id: Mapped[int] = mapped_column(ForeignKey("courses.id"), primary_key=True)
        enrolled_at: Mapped[datetime] = mapped_column(DateTime)
        grade: Mapped[Optional[float]] = mapped_column(Numeric(5, 2))

    class StudentV2(Base):
        __tablename__ = "students"
        id: Mapped[int] = mapped_column(Integer, primary_key=True)
        name: Mapped[str] = mapped_column(String(80))
        enrollments: Mapped[list["StudentCourse"]] = relationship(
            "StudentCourse", back_populates="student", cascade="all, delete-orphan"
        )
        courses: Mapped[list["Course"]] = relationship(
            "Course", secondary="student_course", viewonly=True,
        )

    class CourseV2(Base):
        __tablename__ = "courses"
        id: Mapped[int] = mapped_column(Integer, primary_key=True)
        title: Mapped[str] = mapped_column(String(200))
        enrollments: Mapped[list["StudentCourse"]] = relationship(
            "StudentCourse", back_populates="course", cascade="all, delete-orphan"
        )

    StudentCourse.student = relationship(StudentV2, back_populates="enrollments")
    StudentCourse.course = relationship(CourseV2, back_populates="enrollments")

自引用关系
~~~~~~~~~~

.. code-block:: python

    class Employee(Base):
        __tablename__ = "employees"
        id: Mapped[int] = mapped_column(Integer, primary_key=True)
        name: Mapped[str] = mapped_column(String(80))
        manager_id: Mapped[Optional[int]] = mapped_column(ForeignKey("employees.id"))

        # 一对多自引用：所有下属
        subordinates: Mapped[list["Employee"]] = relationship(
            "Employee",
            remote_side="Employee.id",   # 指向自己
            back_populates="manager",
        )
        manager: Mapped[Optional["Employee"]] = relationship(
            "Employee",
            remote_side="Employee.id",
            back_populates="subordinates",
        )

    # 查所有上级链
    with Session(engine) as s:
        e = s.query(Employee).filter_by(name="小李").one()
        for mgr in [e.manager, e.manager.manager, e.manager.manager.manager]:
            if mgr: print(mgr.name)

继承映射（Inheritance Mapping）
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

.. code-block:: python

    from pbatis.orm import declared_attr

    # 单表继承（joined / concrete / single-table 三种）
    class Person(Base):
        __tablename__ = "people"
        id: Mapped[int] = mapped_column(Integer, primary_key=True)
        name: Mapped[str] = mapped_column(String(80))
        type: Mapped[str] = mapped_column(String(20))   # 鉴别列

        __mapper_args__ = {
            "polymorphic_on": "type",
            "polymorphic_identity": "person",
        }

    class Engineer(Person):
        __tablename__ = "engineers"
        id: Mapped[int] = mapped_column(ForeignKey("people.id"), primary_key=True)
        language: Mapped[str] = mapped_column(String(40))

        __mapper_args__ = {
            "polymorphic_identity": "engineer",
        }

Eager Loading 策略
~~~~~~~~~~~~~~~~~~

.. code-block:: python

    from pbatis.orm import (
        selectinload, joinedload, subqueryload, contains_eager,
    )

    # selectinload：推荐，1 条额外 IN 查询
    stmt = select(User).options(selectinload(User.addresses))

    # joinedload：1 条 LEFT JOIN（适合 many-to-one / one-to-one）
    stmt = select(Address).options(joinedload(Address.user))

    # subqueryload：1 条子查询
    stmt = select(User).options(subqueryload(User.orders))

    # contains_eager：手写 JOIN 但仍按 relationship 填充
    stmt = (
        select(User)
        .join(Address)
        .where(Address.city == "深圳")
        .options(contains_eager(User.addresses))
    )

Session 与生命周期
~~~~~~~~~~~~~~~~~~

.. code-block:: python

    from pbatis.orm import Session, sessionmaker

    # 1. 直接用 Session 类（每次创建）
    with Session(engine) as s:
        s.add(user)
        s.commit()

    # 2. 用 sessionmaker 工厂（推荐，绑定配置）
    SessionLocal = sessionmaker(engine, expire_on_commit=False)

    with SessionLocal() as s:
        s.add(user)
        s.commit()

    # 3. scoped_session（线程/请求级单例）
    from pbatis.orm import scoped_session
    SessionScoped = scoped_session(SessionLocal)
    s = SessionScoped()         # 同线程复用同一 Session
    s.add(user)
    s.commit()
    SessionScoped.remove()      # 结束线程时清理

    # 4. 异步 Session
    from pbatis.ext.asyncio import async_sessionmaker, AsyncSession
    AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
    async with AsyncSessionLocal() as s:
        s.add(user)
        await s.commit()

查询 API
~~~~~~~~

.. code-block:: python

    # Query 1.x 风格（保留兼容）
    users = s.query(User).filter(User.age > 18).order_by(User.id).all()
    one_user = s.query(User).filter_by(email="a@x.com").first()

    # 2.0 风格（推荐）：select() + Session.execute()
    from pbatis import select
    stmt = select(User).where(User.age > 18)
    for u in s.execute(stmt).scalars():
        print(u.name)

    # 复杂条件
    from pbatis import and_, or_, not_
    stmt = select(User).where(
        and_(
            User.age > 18,
            or_(User.city == "深圳", User.city == "广州"),
            not_(User.name.like("test%")),
        )
    )

----

数据库方言（Dialects）
----------------------

PyBatis 通过 URL 中的 ``dialect+driver://`` 形式选择数据库后端与驱动。

URL 格式
~~~~~~~~

.. code-block:: text

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

参数说明：

- ``dialect``：数据库类型（``sqlite``、``mysql``、``postgresql``、``oracle``、``mssql``）
- ``driver``：驱动名称（小写，可选，省略时使用默认驱动）
- 查询参数：作为连接参数透传给驱动

SQLite
~~~~~~

.. code-block:: python

    # 内存库（测试用）
    engine = create_engine("sqlite:///:memory:")

    # 内存共享（多连接可见）
    engine = create_engine("sqlite:///file:memdb1?mode=memory&cache=shared&uri=true")

    # 文件库
    engine = create_engine("sqlite:///data/app.db")
    engine = create_engine("sqlite:////absolute/path/app.db")  # Unix 绝对路径

    # 异步（aiosqlite）
    engine = create_async_engine("sqlite+aiosqlite:///data/app.db")

    # 加密（SQLCipher，需安装 sqlcipher3-binary）
    engine = create_engine("sqlite+pysqlcipher://:password@/data/encrypted.db")

    # 特殊方言：pysqlite_numeric、pysqlite_dollar
    engine = create_engine("sqlite+pysqlite_numeric:///:memory:")  # NUMERIC 精度修复
    engine = create_engine("sqlite+pysqlite_dollar:///:memory:")   # $ 字符串引用

MySQL / MariaDB
~~~~~~~~~~~~~~~

.. code-block:: python

    # PyMySQL（纯 Python，最常用）
    engine = create_engine("mysql+pymysql://user:pwd@localhost:3306/mydb?charset=utf8mb4")

    # mysqlclient（性能好，需 C 扩展）
    engine = create_engine("mysql+mysqldb://user:pwd@localhost:3306/mydb?charset=utf8mb4")

    # mysql-connector-python（Oracle 官方）
    engine = create_engine("mysql+mysqlconnector://user:pwd@localhost:3306/mydb")

    # MariaDB 官方驱动
    engine = create_engine("mariadb+mariadbconnector://user:pwd@localhost:3306/mydb")

    # 异步
    engine = create_async_engine("mysql+aiomysql://user:pwd@localhost:3306/mydb")
    engine = create_async_engine("mysql+asyncmy://user:pwd@localhost:3306/mydb")

    # 多主机（读写分离）
    engine = create_engine("mysql+pymysql://user:pwd@host1,host2/mydb")

    # 常用参数
    #   ?charset=utf8mb4    字符集
    #   ?ssl_ca=/path/ca.pem SSL 证书
    #   ?ssl_verify_cert=false 跳过证书验证

PostgreSQL
~~~~~~~~~~

.. code-block:: python

    # psycopg 3（推荐）
    engine = create_engine("postgresql+psycopg://user:pwd@localhost:5432/mydb")

    # psycopg2-binary
    engine = create_engine("postgresql+psycopg2://user:pwd@localhost:5432/mydb")

    # pg8000（纯 Python）
    engine = create_engine("postgresql+pg8000://user:pwd@localhost:5432/mydb")

    # 异步
    engine = create_async_engine("postgresql+asyncpg://user:pwd@localhost:5432/mydb")

    # PG 特有：ARRAY / JSONB / HSTORE
    from pbatis.dialects.postgresql import ARRAY, JSONB, HSTORE
    from pbatis.dialects.postgresql import insert as pg_insert

    t = Table("t", m, Column("tags", ARRAY(String)), Column("attrs", JSONB))

    # ON CONFLICT (UPSERT)
    stmt = pg_insert(t).values(id=1, name="x").on_conflict_do_update(
        index_elements=[t.c.id], set_={"name": "y"},
    )

    # JSONB 操作
    from pbatis.dialects.postgresql import JSONB
    stmt = select(t).where(t.c.attrs["key"].astext == "value")
    stmt = select(t).where(t.c.attrs.contains({"k": "v"}))

    # 全文搜索
    from pbatis.dialects.postgresql import TSVECTOR
    stmt = select(t).where(t.c.doc.op("@@")(func.to_tsquery("python & sql")))

Oracle
~~~~~~

.. code-block:: python

    # oracledb（推荐，替代旧的 cx_Oracle）
    engine = create_engine("oracle+oracledb://user:pwd@localhost:1521/?service_name=FREEPDB1")

    # cx_Oracle
    engine = create_engine("oracle+cx_oracle://user:pwd@localhost:1521/?service_name=FREEPDB1")

    # TNS 名称
    engine = create_engine("oracle+oracledb://user:pwd@MY_TNS_ALIAS")

    # Oracle 特有类型
    from pbatis.dialects.oracle import RAW, NUMBER, CLOB, BLOB, ROWID

Microsoft SQL Server
~~~~~~~~~~~~~~~~~~~~

.. code-block:: python

    # pyodbc + ODBC Driver 17/18
    engine = create_engine(
        "mssql+pyodbc://user:pwd@localhost:1433/mydb"
        "?driver=ODBC+Driver+18+for+SQL+Server"
        "&TrustServerCertificate=yes"
    )

    # pymssql
    engine = create_engine("mssql+pymssql://user:pwd@localhost:1433/mydb")

    # mssql-python（微软官方纯 Python）
    engine = create_engine("mssql+mssqlpython://user:pwd@localhost:1433/mydb?Encrypt=No")

    # 异步
    engine = create_async_engine("mssql+aioodbc://user:pwd@localhost:1433/mydb?driver=ODBC+Driver+18+for+SQL+Server")

    # MSSQL 特有：IDENTITY、ROWVERSION
    from pbatis.dialects.mssql import ROWVERSION
    t = Table("t", m, Column("id", Integer, primary_key=True), Column("version", ROWVERSION))

方言能力矩阵
~~~~~~~~~~~~

.. list-table::
   :header-rows: 1
   :widths: 15 12 12 12 12 12 12

   * - 能力
     - SQLite
     - MySQL
     - PostgreSQL
     - Oracle
     - MSSQL
     - SQLCipher
   * - 基础 DDL/DML
     - ✓
     - ✓
     - ✓
     - ✓
     - ✓
     - ✓
   * - RETURNING
     - ✓
     - ✗
     - ✓
     - ✓ (21c+)
     - ✓ (OUTPUT)
     - ✓
   * - UPSERT
     - ON CONFLICT
     - ON DUPLICATE KEY
     - ON CONFLICT
     - MERGE
     - MERGE
     - ON CONFLICT
   * - JSON 字段
     - JSON
     - JSON
     - JSONB / JSON
     - CLOB+CHECK
     - NVARCHAR+ISJSON
     - JSON
   * - 数组
     - ✗
     - ✗
     - ✓
     - VARRAY
     - ✗
     - ✗
   * - 全文搜索
     - FTS5
     - FULLTEXT
     - tsvector
     - TextIndex
     - Full-Text
     - FTS5
   * - 窗口函数
     - ✓
     - ✓
     - ✓
     - ✓
     - ✓
     - ✓
   * - CTE
     - ✓
     - ✓
     - ✓
     - ✓
     - ✓
     - ✓
   * - 异步驱动
     - aiosqlite
     - aiomysql / asyncmy
     - asyncpg
     - ✗
     - aioodbc
     - aiosqlite

----

扩展模块（PyBatis Ext）
------------------------

.. code-block:: python

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

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

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

    # hybrid —— 同时在 Python 层与 SQL 层可用的属性
    from pbatis.ext.hybrid import hybrid_property, hybrid_method
    class Product(Base):
        @hybrid_property
        def discounted_price(self):
            return self.price * 0.9
        @discounted_price.expression
        def discounted_price(cls):
            return cls.price * 0.9

    # mutable —— 可变类型变更跟踪
    from pbatis.ext.mutable import MutableDict, MutableList
    class User(Base):
        preferences: Mapped[dict] = mapped_column(MutableDict.as_mutable(JSON))

    # orderinglist —— 有序集合
    from pbatis.ext.orderinglist import ordering_list
    class Playlist(Base):
        tracks: Mapped[list["Track"]] = relationship(
            "Track", order_by="Track.position",
            collection_class=ordering_list("position"),
        )

    # serializer —— pickle 序列化大字段
    from pbatis.ext.serializer import PickleType
    blobs: Mapped[dict] = mapped_column(PickleType)

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

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

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

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

----

扩展模块：雪花 ID（Snowflake）
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

.. code-block:: python

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

ID 结构
^^^^^^^^

64 位二进制按 BigInteger 存储：

.. list-table::
   :header-rows: 1
   :widths: 8 42 8 42

   * - 位
     - 含义
     - 位
     - 含义
   * - 63
     - 符号位（始终 0）
     - 22-12
     - worker_id（5 bit，0-31）
   * - 62-22
     - 时间戳（41 bit，相对 epoch 的毫秒数）
     - 11-0
     - sequence（12 bit，每毫秒 4096 个）

默认 epoch = 1288834974657（Twitter 原版，2010-11-04 01:42:54 UTC）。
41 位时间戳可用约 69 年；5+5 位最多 1024 节点；12 位序列号单节点每毫秒 4096 个 ID。
集群 1024 节点理论峰值约每秒 40 亿 ID。

直接生成 ID
^^^^^^^^^^^^

.. code-block:: python

    from pbatis.ext.snowflake import snowflake_id, SnowflakeGenerator

    # 进程默认生成器
    uid = snowflake_id()       # 例如 1537200202186752
    uid2 = snowflake_id()      # 比 uid 大（趋势递增）

    # 自定义生成器
    gen = SnowflakeGenerator(worker_id=1, datacenter_id=1)
    uid = gen.next_id()

    # 解析 ID（拆出时间戳、节点、序列号）
    info = gen.parse(uid)
    info["datetime"]    # 生成时刻（UTC datetime）
    info["worker_id"]   # 节点 ID
    info["datacenter_id"]
    info["sequence"]    # 毫秒内序列号
    info["timestamp"]   # Unix 毫秒时间戳

用作主键（Core API）
^^^^^^^^^^^^^^^^^^^^

.. code-block:: python

    from pbatis import Table, Column, MetaData, String
    from pbatis.ext.snowflake import SnowflakeID, snowflake_id

    users = Table("users", MetaData(),
        Column("id", SnowflakeID, primary_key=True, default=snowflake_id),
        Column("name", String(50)),
    )

    # INSERT 时自动生成主键
    with engine.begin() as conn:
        r = conn.execute(users.insert().values(name="张三"))
        print(r.inserted_primary_key)   # (2109665447048642560,)

        # 批量 INSERT 也会逐行生成不同 ID
        conn.execute(users.insert(), [
            {"name": "李四"},
            {"name": "王五"},
        ])

    # 也可指定专用生成器
    from pbatis.ext.snowflake import SnowflakeGenerator
    custom_gen = SnowflakeGenerator(worker_id=2, datacenter_id=1)
    special = Table("special", MetaData(),
        Column("id", SnowflakeID, primary_key=True, default=custom_gen.next_id),
        Column("payload", String(100)),
    )

工厂函数（更短）
^^^^^^^^^^^^^^^^

.. code-block:: python

    from pbatis.ext.snowflake import snowflake_column

    # 等价于 Column("id", SnowflakeID, primary_key=True, default=snowflake_id)
    t = Table("t", MetaData(),
        snowflake_column("id", primary_key=True),
        snowflake_column("user_id"),          # 非主键也可用
        Column("data", String(100)),
    )

ORM 用法
^^^^^^^^

.. code-block:: python

    from pbatis.orm import Base, Mapped, mapped_column, Session
    from pbatis.ext.snowflake import SnowflakeID, snowflake_id

    class User(Base):
        __tablename__ = "users"
        id: Mapped[int] = mapped_column(SnowflakeID, primary_key=True, default=snowflake_id)
        name: Mapped[str]

    with Session(engine) as s:
        u = User(name="赵六")
        s.add(u)
        s.commit()
        print(u.id)     # 自动生成

进程级默认生成器
^^^^^^^^^^^^^^^^

.. code-block:: python

    from pbatis.ext.snowflake import (
        configure_default, get_default_generator, set_default_generator,
    )

    # 应用启动时配置一次
    import os
    configure_default(
        worker_id=int(os.environ.get("WORKER_ID", "0")),
        datacenter_id=int(os.environ.get("DC_ID", "0")),
    )

    # 全局未指定 generator 的 SnowflakeID 都用此默认
    from pbatis.ext.snowflake import snowflake_id
    uid = snowflake_id()    # 使用默认生成器

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

线程安全
^^^^^^^^

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

.. code-block:: python

    import threading
    from pbatis.ext.snowflake import snowflake_id

    ids = []
    def worker():
        for _ in range(1000):
            ids.append(snowflake_id())
    threads = [threading.Thread(target=worker) for _ in range(8)]
    for t in threads: t.start()
    for t in threads: t.join()
    assert len(set(ids)) == 8000    # 全部唯一

时钟回拨处理
^^^^^^^^^^^^

.. code-block:: python

    # 容忍 ≤ 5ms 的回拨（自旋等待追上）
    # 超出则抛 ClockMovedBackwardsError
    from pbatis.ext.snowflake import SnowflakeGenerator, ClockMovedBackwardsError

    gen = SnowflakeGenerator(worker_id=0, datacenter_id=0)
    try:
        gen.next_id()
    except ClockMovedBackwardsError as e:
        # 时钟回拨超过 5ms；记录告警 + 等待 + 重试
        logger.error("Clock skew detected: %s", e)
        time.sleep(0.1)
        gen.next_id()

ID 业务化（带业务前缀）
~~~~~~~~~~~~~~~~~~~~~~~~~

如需带业务前缀（如 ``order_``、``user_``），不要直接修改 SnowflakeGenerator，而是封装一个字段：

.. code-block:: python

    from pbatis import Table, Column, MetaData, String
    from pbatis.ext.snowflake import SnowflakeID, snowflake_id

    # 主表存原始 bigint 雪花 ID
    orders = Table("orders", MetaData(),
        Column("id", SnowflakeID, primary_key=True, default=snowflake_id),
        Column("biz_id", String(30), default=lambda: f"ord_{snowflake_id()}"),
        Column("amount", String(20)),
    )

----

配置与连接池
------------

.. code-block:: python

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

    # 只打印部分
    import logging
    logging.basicConfig()
    logging.getLogger("pbatis.engine").setLevel(logging.INFO)

    # 池配置
    engine = create_engine(
        "postgresql+psycopg://...",
        pool_size=10,
        max_overflow=20,
        pool_timeout=30,        # 取连接等待秒数
        pool_recycle=3600,      # 多久回收连接
        pool_pre_ping=True,     # 取出时验证连接有效
    )

    # 关闭连接池（每次新建）
    engine = create_engine("sqlite:///:memory:", poolclass=NullPool)

    # 静态池（所有连接共享同一物理连接，适合内存 SQLite）
    engine = create_engine("sqlite:///:memory:", poolclass=StaticPool, connect_args={"check_same_thread": False})

    # 异步连接池
    engine = create_async_engine(
        "postgresql+asyncpg://...",
        pool_size=5,
        max_overflow=10,
    )

    # 关闭引擎
    engine.dispose()

事件系统
--------

.. code-block:: python

    from pbatis import event

    # 监听连接建立
    @event.listens_for(engine, "connect")
    def receive_connect(dbapi_connection, connection_record):
        cursor = dbapi_connection.cursor()
        cursor.execute("SET time_zone = '+08:00'")

    # 监听连接 checkout（取出）
    @event.listens_for(engine, "checkout")
    def receive_checkout(dbapi_connection, connection_record, connection_proxy):
        pass

    # 监听 ORM 事件
    from pbatis.orm import Session
    @event.listens_for(Session, "before_commit")
    def before_commit(session):
        pass

    @event.listens_for(User, "load")
    def receive_load(target, context):
        pass

----

类型与自定义
------------

.. code-block:: python

    from pbatis import TypeDecorator, String

    # 自定义类型：加密字段
    class EncryptedString(TypeDecorator):
        impl = String
        cache_ok = True

        def process_bind_param(self, value, dialect):
            return encrypt(value) if value else value

        def process_result_value(self, value, dialect):
            return decrypt(value) if value else value

    # 自定义类型：JSON 在 SQLite/MySQL 走 TEXT，在 PG 走 JSONB
    class JsonField(TypeDecorator):
        impl = String
        cache_ok = True

        def load_dialect_impl(self, dialect):
            if dialect.name == "postgresql":
                from pbatis.dialects.postgresql import JSONB
                return dialect.type_descriptor(JSONB())
            return dialect.type_descriptor(String())

    # 枚举字段
    import enum
    class Color(enum.Enum):
        RED = "red"
        GREEN = "green"
        BLUE = "blue"

    Column("color", Enum(Color))

----

反射（Reflection）
------------------

从已有数据库读取表结构，无需手写 Table 定义：

.. code-block:: python

    from pbatis import MetaData, Table, create_engine, inspect

    engine = create_engine("postgresql+psycopg://localhost/mydb")
    metadata = MetaData()

    # 反射单张表
    users = Table("users", metadata, autoload_with=engine)
    print(users.c)

    # 反射所有表
    metadata.reflect(bind=engine)
    for table in metadata.sorted_tables:
        print(table.name)

    # 用 inspect API
    inspector = inspect(engine)
    print(inspector.get_table_names())
    print(inspector.get_columns("users"))
    print(inspector.get_foreign_keys("orders"))
    print(inspector.get_indexes("users"))
    print(inspector.get_pk_constraint("users"))
    print(inspector.get_unique_constraints("users"))
    print(inspector.get_check_constraints("users"))

----

迁移与版本管理
--------------

PyBatis 不内置 schema 迁移工具；推荐使用外部迁移工具（业内常用 `Alembic`）配合 PyBatis：

.. code-block:: bash

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

----

使用示例
--------

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

完整博客系统（端到端）
~~~~~~~~~~~~~~~~~~~~~~

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

.. code-block:: python

    from datetime import datetime
    from pbatis import (
        create_engine, MetaData, Table, Column, Integer, String, Text,
        ForeignKey, DateTime, Boolean, select, insert, update, delete,
        func, and_, or_,
    )
    from pbatis.orm import (
        declarative_base, Mapped, mapped_column, relationship,
        Session, sessionmaker, selectinload,
    )

    # ---------- 1. Schema 定义（Core API）----------
    metadata = MetaData()

    users = Table("users", metadata,
        Column("id", Integer, primary_key=True, autoincrement=True),
        Column("username", String(50), unique=True, nullable=False),
        Column("email", String(120), unique=True, nullable=False),
        Column("password_hash", String(128), nullable=False),
        Column("bio", Text),
        Column("created_at", DateTime, server_default=func.now()),
    )

    posts = Table("posts", metadata,
        Column("id", Integer, primary_key=True, autoincrement=True),
        Column("author_id", Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False),
        Column("title", String(200), nullable=False),
        Column("body", Text, nullable=False),
        Column("published", Boolean, default=False),
        Column("view_count", Integer, default=0),
        Column("created_at", DateTime, server_default=func.now()),
        Column("updated_at", DateTime, onupdate=func.now()),
    )

    comments = Table("comments", metadata,
        Column("id", Integer, primary_key=True, autoincrement=True),
        Column("post_id", Integer, ForeignKey("posts.id", ondelete="CASCADE"), nullable=False),
        Column("user_id", Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False),
        Column("content", Text, nullable=False),
        Column("created_at", DateTime, server_default=func.now()),
    )

    # 关联表：多对多（文章 <-> 标签）
    post_tags = Table("post_tags", metadata,
        Column("post_id", Integer, ForeignKey("posts.id", ondelete="CASCADE"), primary_key=True),
        Column("tag_id", Integer, ForeignKey("tags.id", ondelete="CASCADE"), primary_key=True),
    )

    tags = Table("tags", metadata,
        Column("id", Integer, primary_key=True, autoincrement=True),
        Column("name", String(50), unique=True, nullable=False),
        Column("slug", String(60), unique=True, nullable=False),
    )

    # ---------- 2. 引擎与会话工厂 ----------
    engine = create_engine("sqlite:///:memory:", echo=False)
    metadata.create_all(engine)
    SessionLocal = sessionmaker(engine, expire_on_commit=False)

    # ---------- 3. 增 ----------
    with SessionLocal() as s:
        # 用户
        s.execute(insert(users).values(
            username="alice", email="alice@x.com", password_hash="hash1", bio="pythonista"
        ))
        s.execute(insert(users).values(
            username="bob", email="bob@x.com", password_hash="hash2"
        ))

        # 文章
        s.execute(insert(posts).values(
            author_id=1, title="PyBatis 入门",
            body="PyBatis 是一个 Python ORM 库", published=True, view_count=100
        ))
        s.execute(insert(posts).values(
            author_id=1, title="高级用法",
            body="事务、锁、连接池", published=False, view_count=10
        ))

        # 标签 + 多对多
        s.execute(insert(tags).values(name="Python", slug="python"))
        s.execute(insert(tags).values(name="ORM", slug="orm"))
        s.execute(insert(post_tags).values(post_id=1, tag_id=1))
        s.execute(insert(post_tags).values(post_id=1, tag_id=2))
        s.execute(insert(post_tags).values(post_id=2, tag_id=1))

        # 评论
        s.execute(insert(comments).values(
            post_id=1, user_id=2, content="好文！"
        ))
        s.commit()

    # ---------- 4. 查 ----------
    with SessionLocal() as s:
        # 全部已发布文章
        stmt = select(posts).where(posts.c.published == True).order_by(posts.c.view_count.desc())
        for row in s.execute(stmt):
            print(f"  [{row.id}] {row.title} (浏览 {row.view_count})")

        # 某用户全部文章 + 评论数
        stmt = (
            select(
                posts,
                func.count(comments.c.id).label("comment_cnt"),
            )
            .outerjoin(comments, comments.c.post_id == posts.c.id)
            .where(posts.c.author_id == 1)
            .group_by(posts.c.id)
            .order_by(posts.c.id)
        )
        for row in s.execute(stmt):
            print(f"  [{row.id}] {row.title} - {row.comment_cnt} 条评论")

        # 标签为 Python 的文章
        stmt = (
            select(posts)
            .join(post_tags, post_tags.c.post_id == posts.c.id)
            .join(tags, tags.c.id == post_tags.c.tag_id)
            .where(tags.c.slug == "python")
        )
        for row in s.execute(stmt):
            print(f"  Tag 'python': [{row.id}] {row.title}")

    # ---------- 5. 改 ----------
    with SessionLocal() as s:
        # 发布某文章 + 浏览数 +1
        s.execute(
            update(posts)
            .where(posts.c.id == 2)
            .values(published=True, view_count=posts.c.view_count + 1)
        )
        s.commit()

    # ---------- 6. 删 ----------
    with SessionLocal() as s:
        # 删除某评论
        s.execute(delete(comments).where(comments.c.id == 1))
        s.commit()

    # ---------- 7. ORM 模型版（更易维护）----------
    Base = declarative_base()

    class User(Base):
        __tablename__ = "users"
        id: Mapped[int] = mapped_column(primary_key=True)
        username: Mapped[str] = mapped_column(String(50), unique=True)
        email: Mapped[str] = mapped_column(String(120), unique=True)
        posts: Mapped[list["Post"]] = relationship(
            "Post", back_populates="author", cascade="all, delete-orphan"
        )
        comments: Mapped[list["Comment"]] = relationship(
            "Comment", back_populates="user", cascade="all, delete-orphan"
        )

    class Post(Base):
        __tablename__ = "posts"
        id: Mapped[int] = mapped_column(primary_key=True)
        author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
        title: Mapped[str] = mapped_column(String(200))
        body: Mapped[str] = mapped_column(Text)
        published: Mapped[bool] = mapped_column(default=False)
        view_count: Mapped[int] = mapped_column(default=0)
        author: Mapped["User"] = relationship("User", back_populates="posts")
        comments: Mapped[list["Comment"]] = relationship(
            "Comment", back_populates="post", cascade="all, delete-orphan"
        )
        tags: Mapped[list["Tag"]] = relationship(
            "Tag", secondary="post_tags", back_populates="posts"
        )

    class Comment(Base):
        __tablename__ = "comments"
        id: Mapped[int] = mapped_column(primary_key=True)
        post_id: Mapped[int] = mapped_column(ForeignKey("posts.id"))
        user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
        content: Mapped[str] = mapped_column(Text)
        post: Mapped["Post"] = relationship("Post", back_populates="comments")
        user: Mapped["User"] = relationship("User", back_populates="comments")

    class Tag(Base):
        __tablename__ = "tags"
        id: Mapped[int] = mapped_column(primary_key=True)
        name: Mapped[str] = mapped_column(String(50), unique=True)
        slug: Mapped[str] = mapped_column(String(60), unique=True)
        posts: Mapped[list["Post"]] = relationship(
            "Post", secondary="post_tags", back_populates="tags"
        )

    # ---------- 8. ORM 用法（推荐）----------
    with SessionLocal() as s:
        # 查"已发布"文章（预加载作者、标签、评论）
        stmt = (
            select(Post)
            .options(selectinload(Post.author), selectinload(Post.tags))
            .where(Post.published == True)
            .order_by(Post.view_count.desc())
        )
        for post in s.execute(stmt).scalars():
            print(f"  {post.title} by {post.author.username}, 标签={[t.name for t in post.tags]}")

        # 增
        new_post = Post(
            author_id=2, title="ORM 体验", body="...",
            tags=[s.query(Tag).filter_by(slug="orm").one()]
        )
        s.add(new_post)
        s.commit()

        # 改
        post = s.query(Post).filter_by(title="ORM 体验").one()
        post.view_count += 1
        s.commit()

        # 删
        s.query(Comment).filter_by(id=1).delete()
        s.commit()

FastAPI 集成
~~~~~~~~~~~~

PyBatis 2.0 + FastAPI 的标准集成模式：用 Depends 注入 Session。

.. code-block:: python

    # main.py
    from typing import Generator
    from fastapi import Depends, FastAPI, HTTPException
    from pbatis import create_engine
    from pbatis.orm import Session, sessionmaker
    from pbatis.orm import declarative_base, Mapped, mapped_column, Integer, String

    Base = declarative_base()
    engine = create_engine("postgresql+psycopg://user:pwd@localhost/mydb")
    SessionLocal = sessionmaker(engine, expire_on_commit=False, autoflush=False)

    class Item(Base):
        __tablename__ = "items"
        id: Mapped[int] = mapped_column(Integer, primary_key=True)
        name: Mapped[str] = mapped_column(String(100))
        price: Mapped[float] = mapped_column(default=0)

    Base.metadata.create_all(engine)

    app = FastAPI()

    # 依赖：每个请求一个 Session
    def get_db() -> Generator[Session, None, None]:
        db = SessionLocal()
        try:
            yield db
        finally:
            db.close()

    # CRUD 端点
    @app.post("/items")
    def create_item(name: str, price: float, db: Session = Depends(get_db)):
        item = Item(name=name, price=price)
        db.add(item)
        db.commit()
        db.refresh(item)
        return {"id": item.id, "name": item.name}

    @app.get("/items/{item_id}")
    def get_item(item_id: int, db: Session = Depends(get_db)):
        item = db.get(Item, item_id)
        if not item:
            raise HTTPException(404, "item not found")
        return {"id": item.id, "name": item.name, "price": item.price}

    @app.get("/items")
    def list_items(skip: int = 0, limit: int = 20, db: Session = Depends(get_db)):
        from pbatis import select
        stmt = select(Item).offset(skip).limit(limit)
        return [{"id": i.id, "name": i.name, "price": i.price}
                for i in db.execute(stmt).scalars()]

    @app.put("/items/{item_id}")
    def update_item(item_id: int, name: str, price: float, db: Session = Depends(get_db)):
        item = db.get(Item, item_id)
        if not item:
            raise HTTPException(404)
        item.name = name
        item.price = price
        db.commit()
        return {"id": item.id}

    @app.delete("/items/{item_id}")
    def delete_item(item_id: int, db: Session = Depends(get_db)):
        item = db.get(Item, item_id)
        if not item:
            raise HTTPException(404)
        db.delete(item)
        db.commit()
        return {"deleted": item_id}

    # 异步版本（与 FastAPI 异步端点配合）
    from pbatis.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession

    async_engine = create_async_engine("postgresql+asyncpg://user:pwd@localhost/mydb")
    AsyncSessionLocal = async_sessionmaker(async_engine, expire_on_commit=False)

    async def get_async_db() -> AsyncSession:
        async with AsyncSessionLocal() as s:
            yield s

    @app.get("/async-items/{item_id}")
    async def get_async_item(item_id: int, db: AsyncSession = Depends(get_async_db)):
        from pbatis import select
        stmt = select(Item).where(Item.id == item_id)
        result = await db.execute(stmt)
        item = result.scalar_one_or_none()
        if not item:
            raise HTTPException(404)
        return {"id": item.id, "name": item.name}

Flask 集成
~~~~~~~~~~

.. code-block:: python

    # app.py
    from flask import Flask, g
    from pbatis import create_engine
    from pbatis.orm import sessionmaker, Session

    app = Flask(__name__)
    engine = create_engine("postgresql+psycopg://user:pwd@localhost/mydb")
    SessionLocal = sessionmaker(engine)

    @app.before_request
    def create_session():
        g.db = SessionLocal()

    @app.teardown_request
    def close_session(exception=None):
        db = g.pop("db", None)
        if db is not None:
            db.close()

    @app.route("/items")
    def list_items():
        from pbatis import select
        from myapp.models import Item   # 你的 ORM 模型
        items = g.db.execute(select(Item).limit(20)).scalars().all()
        return {"items": [{"id": i.id, "name": i.name} for i in items]}

    @app.route("/items/<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 之外的存储。

.. code-block:: python

    # 在 Django settings.py 配置
    DATABASES = {
        "default": {
            "ENGINE": "django.db.backends.postgresql",
            "NAME": "mydb",
            "USER": "user",
            "PASSWORD": "pwd",
            "HOST": "localhost",
        }
    }

    # myapp/db.py —— PyBatis 引擎单例
    import os
    from pbatis import create_engine
    from pbatis.orm import sessionmaker, declarative_base

    DB_URL = "postgresql+psycopg://{USER}:{PASSWORD}@{HOST}/{NAME}".format(**os.environ)
    engine = create_engine(DB_URL, pool_pre_ping=True)
    SessionLocal = sessionmaker(engine)
    Base = declarative_base()

    # myapp/views.py
    from django.http import JsonResponse
    from myapp.db import SessionLocal
    from myapp.pyb_models import Order    # PyBatis 定义的模型

    def order_list(request):
        from pbatis import select
        with SessionLocal() as s:
            orders = s.execute(select(Order).limit(50)).scalars().all()
            return JsonResponse({"orders": [o.to_dict() for o in orders]}, safe=False)

生产部署清单
~~~~~~~~~~~~

.. code-block:: python

    # 1. 创建引擎（生产配置）
    engine = create_engine(
        os.environ["DATABASE_URL"],
        pool_size=20,             # 池大小
        max_overflow=10,          # 突发容量
        pool_recycle=3600,        # 1 小时回收
        pool_pre_ping=True,       # 取连接前 ping
        pool_timeout=30,          # 等待超时
    )

    # 2. 进程退出时关闭引擎
    import atexit
    atexit.register(engine.dispose)

    # 3. 优雅关闭时 dispose（如 gunicorn worker exit）
    def on_worker_exit(server, worker):
        engine.dispose()

    # 4. 健康检查
    from pbatis import text
    def health_check() -> bool:
        try:
            with engine.connect() as conn:
                conn.execute(text("SELECT 1"))
            return True
        except Exception:
            return False

    # 5. 多租户（按请求选引擎）
    from typing import Dict
    from pbatis import Engine, create_engine
    _engines: Dict[str, Engine] = {}
    def get_engine(tenant_id: str) -> Engine:
        if tenant_id not in _engines:
            _engines[tenant_id] = create_engine(
                f"postgresql+psycopg://user:pwd@host/tenant_{tenant_id}",
                pool_size=5,
            )
        return _engines[tenant_id]

测试用例
~~~~~~~~

.. code-block:: python

    # 用 pytest + 内存 SQLite 跑 PyBatis 应用测试
    # conftest.py
    import pytest
    from pbatis import create_engine
    from pbatis.orm import sessionmaker
    from myapp.models import Base

    @pytest.fixture
    def engine():
        eng = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(eng)
        yield eng
        eng.dispose()

    @pytest.fixture
    def session(engine):
        SessionLocal = sessionmaker(engine)
        s = SessionLocal()
        yield s
        s.close()

    # test_users.py
    from pbatis import select
    from myapp.models import User

    def test_create_user(session):
        u = User(username="alice", email="a@x.com")
        session.add(u)
        session.commit()
        assert u.id is not None

    def test_query_user(session):
        u = session.execute(select(User).where(User.username == "alice")).scalar_one()
        assert u.email == "a@x.com"

----

性能调优
--------

从连接池、查询、ORM、批量操作、流式读取 5 个层面系统讲解如何榨干 PyBatis 性能。

启用 Cython 扩展
~~~~~~~~~~~~~~~~

最快的速度提升：编译 C 扩展，常见操作（行构造、类型转换、缓存键、结果集迭代）从纯 Python 切换到 C 实现，平均 30%-200% 提升。

.. code-block:: bash

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

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

.. code-block:: python

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

连接池调优
~~~~~~~~~~

.. code-block:: python

    # 1. 池大小
    # 经验值：CPU * 2-4 倍（IO 密集型可更大）
    engine = create_engine(
        "postgresql+psycopg://...",
        pool_size=20,
        max_overflow=10,           # 突发容量
        pool_timeout=30,           # 取连接等待秒
    )

    # 2. 回收策略
    engine = create_engine(
        "postgresql+psycopg://...",
        pool_recycle=3600,         # 1 小时回收（避免数据库主动断开）
        pool_pre_ping=True,        # 取连接前 ping（防 stale 连接）
    )

    # 3. 不同场景的池
    # 单进程单请求测试：NullPool
    engine = create_engine(url, poolclass=NullPool)

    # 内存 SQLite：StaticPool 共享同一连接
    engine = create_engine(
        "sqlite:///:memory:",
        poolclass=StaticPool,
        connect_args={"check_same_thread": False},
    )

    # 异步引擎池
    from pbatis.ext.asyncio import create_async_engine
    eng = create_async_engine("postgresql+asyncpg://...", pool_size=10, max_overflow=20)

避免 N+1 查询
~~~~~~~~~~~~~

N+1 是 ORM 最常见的性能陷阱：1 条主查询 + N 条子查询。

.. code-block:: python

    # 错：触发 N+1（10 个 author → 11 条 SQL）
    posts = s.execute(select(Post)).scalars().all()
    for p in posts:
        print(p.author.username)   # 每条都触发一次 SELECT

    # 对 1：selectinload（推荐，1 条额外 IN 查询）
    from pbatis.orm import selectinload
    stmt = select(Post).options(selectinload(Post.author))
    posts = s.execute(stmt).scalars().all()
    for p in posts:
        print(p.author.username)   # 共 2 条 SQL

    # 对 2：joinedload（1 条 LEFT JOIN）
    from pbatis.orm import joinedload
    stmt = select(Post).options(joinedload(Post.author))

    # 对 3：subqueryload（1 条子查询）
    from pbatis.orm import subqueryload
    stmt = select(Post).options(subqueryload(Post.author))

    # 多层预加载
    stmt = select(User).options(
        selectinload(User.posts).selectinload(Post.tags),
        selectinload(User.comments),
    )

    # 手写 JOIN 但仍按 relationship 填充
    from pbatis.orm import contains_eager
    stmt = (
        select(User)
        .join(Post)
        .where(Post.published == True)
        .options(contains_eager(User.posts))
    )

只取需要的列
~~~~~~~~~~~~

.. code-block:: python

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

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

    # ORM 也支持列级限定
    from pbatis.orm import defer, deferrable, undefer
    stmt = select(User).options(defer(User.bio))           # 默认不取 bio
    s.execute(stmt).scalar().bio    # 访问时才触发加载

    # 一次性取消 defer
    s.execute(stmt).scalars().all()
    s.execute(select(User).options(undefer(User.bio))).all()  # 这次取

批量操作
~~~~~~~~

INSERT / UPDATE 逐条提交是最常见的慢来源之一。PyBatis 提供 3 个层级的批量能力。

.. code-block:: python

    # 1. Core API executemany（最快，自动用驱动原生批处理）
    with engine.begin() as conn:
        conn.execute(insert(users), [
            {"name": "u1"}, {"name": "u2"}, {"name": "u3"},
            # ... 10000 行
        ])

    # 2. Session.bulk_save_objects（ORM 批量 INSERT）
    session.bulk_save_objects([
        User(username=f"user_{i}") for i in range(10000)
    ])
    session.commit()

    # 3. Session.bulk_insert_mappings（指定列）
    session.bulk_insert_mappings(
        User,
        [{"username": f"u{i}", "email": f"u{i}@x.com"} for i in range(10000)],
    )
    session.commit()

    # 4. Session.bulk_update_mappings（按主键批量 UPDATE）
    session.bulk_update_mappings(User, [
        {"id": 1, "bio": "new bio 1"},
        {"id": 2, "bio": "new bio 2"},
    ])
    session.commit()

    # 性能对比（10 万行 INSERT 到 PG，psycopg）
    #   逐条 + commit:    47 秒
    #   executemany:      1.8 秒
    #   COPY (raw):       0.4 秒（最快但绕过 ORM）

    # 5. PG 的最快方式：用 raw psycopg.copy_from
    import io
    from pbatis import text
    with engine.raw_connection() as raw:
        with raw.cursor() as cur:
            buf = io.StringIO()
            for u in users_data:
                buf.write(f"{u['name']}\t{u['email']}\n")
            buf.seek(0)
            cur.copy_from(buf, "users", columns=("name", "email"))
        raw.commit()

流式读取大数据集
~~~~~~~~~~~~~~~~

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

.. code-block:: python

    # 1. Core API 流式
    with engine.connect() as conn:
        result = conn.execution_options(stream_results=True).execute(
            select(events).order_by(events.c.id)
        )
        for row in result:
            process(row)     # 一次一行

    # 2. yield_per（ORM 批量缓冲）
    with Session(engine) as s:
        stmt = select(Event).execution_options(yield_per=1000)
        for event in s.execute(stmt).scalars():
            process(event)
    # yield_per 1000 意味着每 1000 行从游标取一次

    # 3. 仅取主键再分批
    with Session(engine) as s:
        total = s.execute(select(func.count(Event.id))).scalar()
        for offset in range(0, total, 5000):
            batch = s.execute(
                select(Event).where(Event.id >= offset).limit(5000)
            ).scalars().all()
            process_batch(batch)

Session 配置
~~~~~~~~~~~~

.. code-block:: python

    # 1. expire_on_commit=False（重要）
    # 默认 True：commit 后实例属性过期，再次访问会重新查询
    # 设 False：commit 后属性仍可用，性能更好
    SessionLocal = sessionmaker(engine, expire_on_commit=False)

    # 2. autoflush=False（批量导入时）
    # 默认 True：每次查询前自动 flush pending 对象
    # 批量导入时关闭可避免无谓的 flush
    session = SessionLocal(autoflush=False)
    session.bulk_save_objects([User(username=f"u{i}") for i in range(1000)])
    session.commit()

    # 3. autobegin=False（显式控制事务边界）
    # 不常用

    # 4. 关闭 identity map（不缓存对象）
    session = SessionLocal()
    # 暂未提供原生开关，可改用 bulk_* 系列方法

    # 5. 查询后再 expire 全部对象（释放内存）
    session.expire_all()

    # 6. detach 单个对象
    session.expunge(user)

事务粒度
~~~~~~~~

.. code-block:: python

    # 错：每条 INSERT 一个事务（10000 次 commit = 慢）
    for u in users_data:
        with engine.begin() as conn:
            conn.execute(insert(users).values(**u))

    # 对：1 个事务包 10000 条
    with engine.begin() as conn:
        for u in users_data:
            conn.execute(insert(users).values(**u))

    # 更好：executemany
    with engine.begin() as conn:
        conn.execute(insert(users), users_data)

    # 长事务 vs 短事务
    # 经验：单个事务 1-10 秒；长事务会持有锁、阻塞 vacuum
    # 批处理：每 1000 行 1 个事务（折中）
    BATCH = 1000
    with engine.begin() as conn:
        for i, u in enumerate(users_data):
            conn.execute(insert(users).values(**u))
            if (i + 1) % BATCH == 0:
                # 这里不能用嵌套 begin（savepoint 模式）
                pass

    # 批量提交（多次短事务）
    for i in range(0, len(users_data), BATCH):
        batch = users_data[i:i + BATCH]
        with engine.begin() as conn:
            conn.execute(insert(users), batch)

SQL 层优化
~~~~~~~~~~

.. code-block:: python

    # 1. 索引友好查询
    # 前缀 % 走不到索引 → 全文搜索替代
    select(users).where(users.c.name.like("张%"))         # 走索引
    select(users).where(users.c.name.like("%张"))         # 全表扫描
    # 用 FULLTEXT 替代：tsvector、MySQL MATCH、SQLite FTS5

    # 2. EXISTS 替代 IN（大子查询）
    select(users).where(users.c.id.in_(select(orders.c.user_id)))    # 可能慢
    select(users).where(exists().where(orders.c.user_id == users.c.id))  # 短路

    # 3. UNION ALL 替代 UNION（不排序去重）
    from pbatis import union_all
    union_all(...)   # 更快

    # 4. 避免 SELECT *
    select(users.c.id, users.c.name)         # OK
    select(users)                            # 全列

    # 5. 配合覆盖索引
    # 如果查询只需要 (a, b)，且有索引 idx_c(a, b, c)，则索引扫描即可
    select(t.c.a, t.c.b) where(t.c.a == 1)  # 走覆盖索引，无须回表

    # 6. 用 EXPLAIN 分析
    from pbatis import text
    plan = s.execute(text("EXPLAIN ANALYZE SELECT * FROM users WHERE name = :n"),
                     {"n": "alice"}).fetchall()
    for row in plan:
        print(row[0])

异步 + 并发
~~~~~~~~~~~

.. code-block:: python

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

    async def fetch_user(s, uid):
        from pbatis import select
        result = await s.execute(select(User).where(User.id == uid))
        return result.scalar_one_or_none()

    async def main():
        eng = create_async_engine("postgresql+asyncpg://...")
        async with eng.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        async with AsyncSession(eng) as s:
            # 并发查 100 个用户
            users = await asyncio.gather(*[fetch_user(s, i) for i in range(1, 101)])
            return users

    # 2. asyncmy 配合 PyMySQL 池
    eng = create_async_engine("mysql+asyncmy://...")

    # 3. asyncpg（PG 最快的异步驱动）
    eng = create_async_engine("postgresql+asyncpg://...")

诊断与监控
~~~~~~~~~~

.. code-block:: python

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

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

    # 3. 慢查询捕获（自定义事件）
    from pbatis import event
    import time

    @event.listens_for(engine, "before_execute")
    def before(conn, clauseelement, multiparams, params, execution_options):
        conn._query_start = time.time()

    @event.listens_for(engine, "after_execute")
    def after(conn, clauseelement, multiparams, params, result, execution_options):
        elapsed = time.time() - conn._query_start
        if elapsed > 0.5:
            logger.warning(f"Slow query ({elapsed:.2f}s): {clauseelement}")

    # 4. 监控：每条 SQL 计数
    from collections import Counter
    counter = Counter()
    @event.listens_for(engine, "after_execute")
    def count(conn, clauseelement, *a, **kw):
        table = clauseelement.table.name if hasattr(clauseelement, "table") else "?"
        counter[table] += 1
    # 查最热表
    print(counter.most_common(10))

    # 5. pg_stat_statements（PG）
    s.execute(text("SELECT * FROM pg_stat_statements ORDER BY total_time DESC LIMIT 10"))

常见性能陷阱
~~~~~~~~~~~~

.. list-table::
   :header-rows: 1
   :widths: 45 55

   * - 陷阱
     - 修法
   * - 触发 N+1（懒加载关联对象）
     - 用 ``selectinload`` / ``joinedload`` 预加载
   * - ``session.query().all()`` 一次加载全表
     - 用 ``yield_per`` 流式 + 批处理
   * - ``expire_on_commit=True`` 重复查询
     - 设 ``expire_on_commit=False``
   * - 逐条 INSERT + 逐条 commit
     - 1 个事务 + executemany
   * - ``SELECT *`` 拉大字段
     - 显式列出列；``defer`` 不常用大字段
   * - 前缀 ``%`` LIKE 用不到索引
     - 用全文搜索 / 反向索引
   * - 每次 query 触发 autoflush
     - 批量导入时 ``autoflush=False``
   * - 长事务持有锁
     - 拆分多个短事务（每 1k 行一个）
   * - ``session.merge()`` 全字段对比
     - 用 ``bulk_update_mappings`` 显式列
   * - 反射时 ``autoload=True`` 启动慢
     - 用 ``reflect()`` 一次性反射

性能基线（参考）
~~~~~~~~~~~~~~~~

在不同场景下，PyBatis 2.x + Cython 大致能达到：

.. list-table::
   :header-rows: 1
   :widths: 30 22 24 24

   * - 操作
     - SQLite (内存)
     - PostgreSQL (psycopg)
     - MySQL (PyMySQL)
   * - 单条 SELECT（已建索引）
     - ~50 µs
     - ~200 µs
     - ~300 µs
   * - 单条 INSERT
     - ~80 µs
     - ~300 µs
     - ~500 µs
   * - 批量 INSERT（1 万行，executemany）
     - ~50 ms
     - ~250 ms
     - ~500 ms
   * - N+1 查询（10 个子对象）
     - 11 次 SQL → 1 次（用 selectinload 后）
     - 同左
     - 同左
   * - ORM 单对象 fetch（懒加载）
     - ~60 µs
     - ~250 µs
     - ~350 µs

*数据为 2024 年中高端笔记本（Apple M2 Pro / 16GB / 本地 PG 14）的参考值。生产环境因网络、并发、索引设计差异可能 ±50%。*

----

常见问题
--------

**Q: ``import pbatis`` 报 ModuleNotFoundError？**

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

**Q: SQLite 文件被占用？**

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

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

设置 ``pbatis_warn_20 = "false"`` 或在环境变量中 ``export PBATIS_WARN_20=false``。

**Q: 性能调优要点？**

1. 用 ``engine.execution_options(stream_results=True)`` 流式读取大数据
2. 批量 ``insert()`` / ``executemany``
3. ORM 懒加载关掉、用 ``selectinload`` / ``joinedload`` 预加载
4. 用 ``session.execute()`` + 2.0 风格 select()，避免 1.x Query 缓存开销
5. 启用 Cython 加速：``pip install cython`` 然后 ``pip install -e .`` 时会自动编译 ``_processors_cy``、``_row_cy``、``_result_cy``、``_cache_key_cy``、``_util_cy``、``_collections_cy``、``_immutabledict_cy`` 等 C 扩展

**Q: 如何关闭所有 SQL 日志？**

.. code-block:: python

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

**Q: 异步与同步能否混用？**

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

----

版本与许可
----------

当前版本：1.1.0

PyPI 包名：**pbatis**；安装命令 ``pip install pbatis``，项目页 https://pypi.org/project/pbatis/

更多资源：

- 完整变更日志：``CHANGELOG.md``
- 一页纸速查：``doc/build/quickref_zh.rst``
- 源码仓库：https://gitee.com/pbatis/pbatis

协议：MIT（详见 ``LICENSE`` 文件）
