Metadata-Version: 2.4
Name: dbqk
Version: 2.1.0
Summary: 一个轻量级的 MySQL / PostgreSQL 连接池 + CRUD 封装库，基于 DBUtils。
Author-email: wauo <markadc@126.com>
License-Expression: MIT
Project-URL: Homepage, https://github.com/markadc/dbqk
Project-URL: Issues, https://github.com/markadc/dbqk/issues
Project-URL: Repository, https://github.com/markadc/dbqk
Keywords: mysql,postgresql,pymysql,psycopg2,dbutils,connection-pool,orm,crud
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Provides-Extra: mysql
Requires-Dist: dbutils>=3.1.0; extra == "mysql"
Requires-Dist: pymysql>=1.1.0; extra == "mysql"
Provides-Extra: pgsql
Requires-Dist: dbutils>=3.1.0; extra == "pgsql"
Requires-Dist: psycopg2-binary>=2.9.0; extra == "pgsql"
Provides-Extra: all
Requires-Dist: dbutils>=3.1.0; extra == "all"
Requires-Dist: pymysql>=1.1.0; extra == "all"
Requires-Dist: psycopg2-binary>=2.9.0; extra == "all"
Dynamic: license-file

# dbqk

[![PyPI](https://img.shields.io/pypi/v/dbqk.svg)](https://pypi.org/project/dbqk/)
[![Python](https://img.shields.io/pypi/pyversions/dbqk.svg)](https://pypi.org/project/dbqk/)
[![License](https://img.shields.io/pypi/l/dbqk.svg)](https://github.com/markadc/dbqk/blob/main/LICENSE)

`dbqk` 是一个同步、轻量的 MySQL / PostgreSQL 连接池与 CRUD 封装库。它基于
[`DBUtils`](https://pypi.org/project/dbutils/)，底层使用 PyMySQL 和 Psycopg2。

## 特性

- DBUtils 连接池和自动连接归还
- MySQL / PostgreSQL 统一的 `Database`、`Table` 和 `ExecResult` API
- 参数化条件、受控标识符和结构化排序
- 默认阻止无条件 `UPDATE` / `DELETE`
- 显式多语句事务，事务内误用 `Database` 会立即报错
- PostgreSQL `INSERT` / `UPDATE` / `DELETE` 的 `RETURNING`
- PostgreSQL 批量插入使用 `execute_values`
- 查询结果支持字典或元组格式
- 按后端安装驱动（`dbqk[mysql]` / `dbqk[pgsql]`），`import dbqk` 不加载驱动

要求 Python 3.10 或更高版本。

## 导航

第一次使用？按你的后端走这两条路径：

- **快速开始**
  - [接入 MySQL 并使用](#接入-mysql-并使用)
  - [接入 PostgreSQL 并使用](#接入-postgresql-并使用)
- **日常使用**
  - [WHERE 条件](#where-条件)
  - [排序与分页](#排序与分页)
  - [全表修改护栏](#全表修改护栏)
  - [事务](#事务)
  - [原生 SQL 与结果对象](#原生-sql-与结果对象)
  - [批量插入](#批量插入)
  - [大表扫描 scan](#大表扫描-scan)
  - [异常](#异常)
- **参考**
  - [标识符规则](#标识符)
  - [从 1.0 迁移到 2.0](#从-10-迁移到-20)
  - [本地开发与测试](#本地开发与测试)

## 接入 MySQL 并使用

```bash
pip install "dbqk[mysql]"
```

```python
from dbqk.mysql_manager import Database

db = Database(
    host="127.0.0.1",
    port=3306,
    user="root",
    password="secret",
    database="app",
)

users = db["users"]
created = users.insert({"name": "Tom", "age": 18})
user_id = created.lastrowid      # MySQL 直接提供自增 ID

rows = users.select(
    where={"age__gte": 18},
    columns=["id", "name", "age"],
    order_by=[("age", "desc"), ("id", "asc")],
).rows

users.update({"age": 19}, where={"id": user_id})
users.delete(where={"id": user_id})
db.close()
```

MySQL 不支持 `returning` 参数（`insert` / `update` / `delete` 均如此），传入会抛出
`NotImplementedError`。

## 接入 PostgreSQL 并使用

```bash
pip install "dbqk[pgsql]"
```

```python
from dbqk.pgsql_manager import Database

db = Database(
    host="127.0.0.1",
    port=5432,
    user="postgres",
    password="secret",
    database="app",
)

users = db["public.users"]
```

PostgreSQL 的普通表不保证驱动级 `lastrowid`。需要拿到插入、更新或删除产生的结果时，
显式使用 `returning`：

```python
result = users.insert({"name": "Tom", "age": 18}, returning="id")
user_id = result.scalar

result = users.insert(
    [
        {"name": "A", "age": 20},
        {"name": "B", "age": 21},
    ],
    returning=["id", "name"],
    page_size=100,
)
inserted_rows = result.rows

returned = users.update({"age": 20}, where={"id": user_id}, returning=["id", "age"])
updated_age = returned.first["age"]

returned = users.delete(where={"id": user_id}, returning="id")
deleted_id = returned.scalar

db.close()
```

同时使用两个后端时安装 `dbqk[all]`。

## 安全查询 API

### 标识符

表名、列名和排序列只接受普通标识符或点分名称，例如 `users`、`public.users`。
每一段必须匹配 `[A-Za-z_][A-Za-z0-9_$]*`。表达式、通配符和任意 SQL 片段不能作为
标识符传入；复杂查询请使用 `db.exec()` 和固定 SQL。

### WHERE 条件

```python
users.select(where={"name": "Tom"})
users.select(where={"age__gt": 18})
users.select(where={"age__gte": 18})
users.select(where={"age__lt": 60})
users.select(where={"age__lte": 60})
users.select(where={"name__ne": "Admin"})
users.select(where={"name__like": "T%"})
users.select(where={"id__in": [1, 2, 3]})       # IN
users.select(where={"id__not_in": [1, 2, 3]})   # NOT IN
users.select(where={"deleted_at": None})       # IS NULL
users.select(where={"deleted_at__ne": None})   # IS NOT NULL
```

支持的后缀为 `eq`、`gt`、`gte`、`lt`、`lte`、`ne`、`like`、`in`、`not_in`。未知后缀会抛出
`ValueError`，不会静默退化为等值条件。多个条件使用 `AND` 连接。

注意：`in` / `not_in` 的值必须是非空列表、元组或集合；列名本身包含 `__` 时无法使用
后缀语法（会被解析成操作符），这类列请改用 `db.exec()` 固定 SQL。

### 排序与分页

```python
users.select(
    order_by=[("created_at", "desc"), ("id", "asc")],
    limit=20,
    offset=40,
)
```

排序方向只允许 `asc` 和 `desc`。为了保持 MySQL / PostgreSQL 行为一致，`offset`
必须与 `limit` 一起使用。

### 全表修改护栏

```python
# 默认拒绝
users.update({"active": False})
users.delete()

# 确认确实需要全表操作时显式授权
users.update({"active": False}, allow_all=True)
users.delete(allow_all=True)
```

## 事务

每个普通 `db.exec()` 或 Table 操作都是独立事务。需要把多个操作组成一个原子单元时，
使用显式事务执行器：

```python
with db.transaction() as tx:
    account = tx["accounts"].find_one({"id": 1}).first
    tx["accounts"].update(
        {"balance": account["balance"] - 100},
        where={"id": 1},
    )
    tx["ledger"].insert({"account_id": 1, "amount": -100})
```

事务正常退出时提交，异常退出时回滚。事务对象绑定单一连接，不能嵌套、重复进入或跨线程
使用；任一 SQL 执行失败后，事务立即变为不可继续使用。

事务进行中直接使用 `db`（例如 `db["users"]` 而不是 `tx["users"]`）会抛出
`TransactionError`——这种写法会从连接池取新连接并立即提交、静默脱离事务，因此被禁止。
确需在事务外独立执行时，使用另一个 `Database` 实例。

## 原生 SQL 与结果对象

```python
result = db.exec(
    "SELECT id, name FROM users WHERE age >= %s",
    [18],
    fetch_mode="dict",  # 或 "tuple"
)

result.rows
result.first
result.scalar
result.one()
result.rowcount
result.lastrowid
result.sql
result.params
result.sql_kind
```

执行器根据游标是否真正产生结果列决定是否读取结果，因此支持 CTE、带前置注释的查询和
PostgreSQL `RETURNING`。设置 `fetch=False` 可显式放弃结果集。

`ExecResult` 支持迭代、索引、`len()` 和布尔判断。PostgreSQL 的 `.lastrowid` 不作保证，
请使用 `returning`。

## 批量插入

```python
result = users.insert([
    {"name": "A", "age": 1},
    {"name": "B", "age": 2},
])
```

所有行必须是非空字典且具有完全相同的列集合。PostgreSQL Table 批量插入使用
`execute_values` 并可通过 `page_size` 控制分批大小。通用 `db.exec(..., many=True)`
仍保留逐条执行语义，以返回准确的受影响行数。

## 大表扫描 scan

`select()` 一次读取全部结果，大表会吃满内存。`scan()` 用键序游标分页（keyset
pagination）分批产出数据，每一批都是索引定位 + 顺序扫描，千万级表也适用：

```python
for rows in users.scan(sort_field="id", batch_size=1000):
    for row in rows:
        process(row)

# 同样支持结构化条件、指定列、批间节流
for rows in users.scan(
    sort_field="id",
    where={"status": 1, "id__gte": 100, "id__lt": 9000},
    columns=["id", "name"],
    batch_size=500,
    rest=0.05,
):
    handle(rows)
```

约束与行为：

- `sort_field` 必须唯一（通常是主键），否则批次边界上的重复值会被跳过；
- 不足一批时扫描自然结束，空表或范围无数据不报错；
- `rest` 是每批之间的休眠秒数，用于控制对数据库的压力（默认 0）；
- `fetch_mode="tuple"` 时必须显式传 `columns`（且包含 `sort_field`），因为元组行
  需要按列位置推进游标；
- 在事务内使用 `tx["users"].scan(...)` 时，整趟扫描共享同一连接，PostgreSQL
  下可以获得一致性快照。

## 异常

```python
from dbqk.pgsql_manager import ExecError, TransactionError

try:
    with db.transaction() as tx:
        tx.exec("UPDATE users SET name = %s WHERE id = %s", ["Tom", 1])
except ExecError:
    ...
except TransactionError:
    ...
```

MySQL 和 PostgreSQL 子包分别导出其后端 `ExecError`，并共同导出
`TransactionError`。

## 从 1.0 迁移到 2.0

- 安装方式变化：`pip install dbqk` 不再默认携带驱动，请按后端安装
  `dbqk[mysql]` / `dbqk[pgsql]` / `dbqk[all]`；缺失驱动时导入会给出明确提示。
- 事务内直接使用 `db`（而非 `tx`）现在抛出 `TransactionError`，不再静默脱离事务。
- `select()` 的 `limit` / `offset` 只接受 `int`；字符串或浮点数不再被隐式转换。
- `select(columns="id")` 这类误传字符串现在抛出 `TypeError`（之前会被逐字符
  当成列名）。
- WHERE 新增 `in` / `not_in` 操作符；`update()` / `delete()` 新增 `returning`
  参数（仅 PostgreSQL）。
- 移除了从未被库抛出的 `TableNotFoundError` 导出。
- 执行失败且回滚也失败时，回滚错误信息会附加在 `ExecError` 消息中。

## 本地开发与测试

```bash
pip install -e ".[all]"
```

测试不依赖任何测试框架，直接运行 Python 脚本。核心检查无需数据库：

```bash
python _tests/run_all.py core
```

统一执行全部脚本时，MySQL 和 PostgreSQL 会直接连接；连接失败或断言失败时脚本返回非零：

```bash
python _tests/run_all.py
```

连接配置集中在各后端的 `_helpers.py`，也可以通过环境变量覆盖后分别执行：

```bash
DBQK_MYSQL_HOST=127.0.0.1 \
DBQK_MYSQL_USER=root \
DBQK_MYSQL_PASSWORD=secret \
DBQK_MYSQL_DATABASE=test \
python _tests/run_all.py mysql

DBQK_PGSQL_HOST=127.0.0.1 \
DBQK_PGSQL_USER=postgres \
DBQK_PGSQL_PASSWORD=secret \
DBQK_PGSQL_DATABASE=test \
python _tests/run_all.py pgsql
```

测试只创建和删除随机命名的 `dbqk_test_<uuid>` 表，不会复用或删除既有业务表。

## License

MIT
