SQL 工具 - SqlUtil¶

由来¶

在日常开发中，手动拼接 SQL 语句容易出错且代码冗长。SqlUtil 提供了常用的 SQL 语句生成方法，支持 SELECT、INSERT、UPDATE、DELETE 以及备份表创建。支持 MySQL、PostgreSQL、SQLite 三种方言。

方言支持¶

所有方法均接受 dialect 参数，可选值为 "mysql"（默认）、"postgresql"、"sqlite"。

特性	MySQL	PostgreSQL	SQLite
标识符引用	`name`	`"name"`	`name`
分页	`LIMIT offset, count`	`LIMIT count OFFSET offset`	`LIMIT count OFFSET offset`
REPLACE INTO	`REPLACE INTO`	`ON CONFLICT DO UPDATE SET`	`INSERT OR REPLACE INTO`
INSERT IGNORE	`INSERT ignore INTO`	`ON CONFLICT DO NOTHING`	`INSERT OR IGNORE INTO`
ON DUPLICATE KEY	`ON DUPLICATE KEY UPDATE`	`ON CONFLICT DO UPDATE SET`	`ON CONFLICT DO UPDATE SET`
VALUES 引用	`VALUES(col)`	`EXCLUDED.col`	`excluded.col`

方法¶

生成查询 SQL¶

from hutool import SqlUtil

# MySQL（默认）
SqlUtil.make_select_sql("user", ["name", "age"])
# 'SELECT `name`,`age` FROM `user`'

# 带条件和排序
SqlUtil.make_select_sql("user", "*", condition="age > 18", order_by=["id"])
# 'SELECT * FROM `user` WHERE age > 18 ORDER BY `id`'

# 分页查询
SqlUtil.make_select_sql("user", "*", page_no=2, page_size=10)
# 'SELECT * FROM `user` LIMIT 10, 10'

# PostgreSQL
SqlUtil.make_select_sql("user", "*", page_no=2, page_size=10, dialect="postgresql")
# 'SELECT * FROM "user" LIMIT 10 OFFSET 10'

# SQLite
SqlUtil.make_select_sql("user", ["name", "age"], dialect="sqlite")
# 'SELECT `name`,`age` FROM `user`'

生成插入 SQL¶

# 单条插入（MySQL）
SqlUtil.make_insert_sql("user", {"name": "Alice", "age": 20})

# 批量插入
data = [{"name": "Alice", "age": 20}, {"name": "Bob", "age": 30}]
SqlUtil.make_insert_sql("user", data)

# INSERT IGNORE（MySQL）
SqlUtil.make_insert_sql("user", {"name": "Alice"}, insert_ignore=True)

# INSERT IGNORE（PostgreSQL → ON CONFLICT DO NOTHING）
SqlUtil.make_insert_sql("user", {"name": "Alice"}, insert_ignore=True, dialect="postgresql")

# INSERT IGNORE（SQLite → INSERT OR IGNORE）
SqlUtil.make_insert_sql("user", {"name": "Alice"}, insert_ignore=True, dialect="sqlite")

# REPLACE INTO（MySQL）
SqlUtil.make_insert_sql("user", {"name": "Alice"}, auto_update=True)

# ON DUPLICATE KEY UPDATE（MySQL）
SqlUtil.make_insert_sql("user", {"name": "Alice", "age": 20}, update_columns=("name",))

# ON CONFLICT DO UPDATE SET（PostgreSQL，需指定 on_conflict）
SqlUtil.make_insert_sql(
    "user", {"name": "Alice", "age": 20}, update_columns=("name",), on_conflict=("id",), dialect="postgresql"
)

# ON CONFLICT DO UPDATE SET（SQLite）
SqlUtil.make_insert_sql(
    "user", {"name": "Alice", "age": 20}, update_columns=("name",), on_conflict=("id",), dialect="sqlite"
)

生成更新 SQL¶

# MySQL
SqlUtil.make_update_sql("user", {"name": "Bob"}, "id = 1")
# "UPDATE `user` SET `name`='Bob' WHERE id = 1;\n"

# PostgreSQL
SqlUtil.make_update_sql("user", {"name": "Bob"}, "id = 1", dialect="postgresql")
# 'UPDATE "user" SET "name"=\'Bob\' WHERE id = 1;\n'

生成删除 SQL¶

SqlUtil.make_delete_sql("user", "id = 1")
# 'DELETE FROM user WHERE id = 1'

创建备份表¶

# 使用默认日期
SqlUtil.create_bak_table_sql("user")
# 'create table user_20240115 as select * from user;\n'

# 指定日期
SqlUtil.create_bak_table_sql("user", bak_date="20240101")
# 'create table user_20240101 as select * from user;\n'

# 批量备份
SqlUtil.create_bak_table_sql(["t1", "t2"], bak_date="20240101")

值格式化¶

SqlUtil._format_sql_value 内部方法支持自动格式化以下类型：

类型	转换
`str`（含 `select`）	用 `${}` 包裹为子查询
`list` / `dict`	转为 JSON 字符串
`datetime.datetime`	`%Y-%m-%d %H:%M:%S`
`datetime.date`	`%Y-%m-%d`
`datetime.time`	`%H:%M:%S`
`bool`	`0` / `1`
numpy 浮点类型	`float()`
numpy 整数类型	`int()`

备注

numpy 支持是可选的。安装了 numpy 时会自动识别 np.float16/32/64、np.int8/16/32/64 等类型并转为 Python 原生类型；未安装 numpy 时完全不影响使用。

链式调用 API¶

除了上述静态方法，SqlUtil 还提供了链式调用 API，通过 SqlUtil.select()、SqlUtil.insert() 等工厂方法创建 Builder 实例，配合 Q（条件组合）、F（字段引用）、ColumnType（列类型枚举）可以链式生成复杂 SQL 语句。

Q — 条件组合器¶

Q 支持 AND(&)、OR(|)、NOT(~) 组合条件。

from hutool import SqlUtil, Q

# 简单条件
Q(name="Alice")  # → `name`='Alice'

# OR
Q(age__gt=18) | Q(role="admin")
# → (`age`>18 OR `role`='admin')

# AND（显式）
Q(status="active") & Q(dept="IT")
# → `status`='active' AND `dept`='IT'

# NOT
~Q(status="deleted")
# → NOT `status`='deleted'

# 复杂嵌套
condition = (Q(age__gte=18, status="active") | Q(role="admin")) & Q(dept__in=["IT", "HR"])

支持的操作符： __gt、__gte、__lt、__lte、__ne、__in、__not_in、__between、__like、__contains、__startswith、__endswith、__isnull

F — 字段引用¶

F 用于字段间比较、算术运算和聚合函数。

from hutool import SqlUtil, F

# 算术运算
F("balance") + 100  # → `balance`+100
F("stock") - 10  # → `stock`-10
F("price") * 1.5  # → `price`*1.5

# 聚合函数
F.count("*")  # → COUNT(*)
F.sum("amount")  # → SUM(`amount`)
F.avg("score")  # → AVG(`score`)
F.max("price")  # → MAX(`price`)
F.min("price")  # → MIN(`price`)

SELECT 链式查询¶

from hutool import SqlUtil, Q, F

# 基础查询
sql = (
    SqlUtil.select("user")
    .columns("id", "name", "age")
    .where(age__gt=18, status="active")
    .order_by("-age", "name")
    .limit(10)
    .offset(20)
    .sql()
)

# 不指定 columns → SELECT *
sql = SqlUtil.select("user").where(id=1).sql()

# Q 组合条件
sql = SqlUtil.select("user").where(Q(age__gt=18) | Q(role="admin")).where(~Q(status="deleted")).sql()

# F 字段引用
sql = SqlUtil.select("product").where(price__gt=F("cost") * 1.5).sql()

# 原始 SQL 片段
sql = SqlUtil.select("user").where("find_in_set(1, status)", Q(age__gt=18)).where_raw("YEAR(created_at) = 2024").sql()

# GROUP BY + HAVING
sql = SqlUtil.select("order").columns("user_id", "COUNT(*) AS cnt").group_by("user_id").having(cnt__gt=5).sql()

# DISTINCT
sql = SqlUtil.select("user").columns("city").distinct().sql()

# 分页
sql = SqlUtil.select("user").columns("*").page(2, 10).sql()

方法列表：

方法	说明
`.columns(*cols)`	指定查询列，可传原始 SQL 表达式
`.where(args, *kwargs)`	添加条件（Q 对象、原始 SQL、field__op=value）
`.where_raw(condition)`	添加原始 SQL WHERE 条件
`.order_by(*fields)`	排序，前缀 `-` 表示 DESC
`.limit(n)` / `.offset(n)`	分页
`.page(page_no, page_size)`	按页码分页
`.distinct()`	去重
`.group_by(*fields)`	分组
`.having(args, *kwargs)`	HAVING 条件

INSERT 链式插入¶

# 基础插入
sql = SqlUtil.insert("user").values(name="Alice", age=20).sql()

# 批量插入
data = [{"name": "Alice", "age": 20}, {"name": "Bob", "age": 30}]
sql = SqlUtil.insert("user").values_list(data).sql()

# INSERT IGNORE
sql = SqlUtil.insert("user").values(name="Alice").ignore().sql()

# REPLACE INTO
sql = SqlUtil.insert("user").values(name="Alice").replace().sql()

# ON DUPLICATE KEY UPDATE（MySQL）
sql = SqlUtil.insert("user").values(name="Alice", age=20).do_update("name").sql()

# ON CONFLICT DO UPDATE SET（PostgreSQL）
sql = (
    SqlUtil.insert("user", dialect="postgresql")
    .values(name="Alice", age=20)
    .on_conflict("id")
    .do_update("name", "age")
    .sql()
)

# ON CONFLICT DO NOTHING
sql = SqlUtil.insert("user", dialect="postgresql").values(name="Alice").do_nothing().sql()

UPDATE 链式更新¶

# 基础更新
sql = SqlUtil.update("user").set(name="Bob", age=30).where(id=1).sql()

# F 字段引用
sql = SqlUtil.update("account").set(balance=F("balance") + 100).where(id=1).sql()

# 原始 SET
sql = SqlUtil.update("account").set_raw("balance = balance + 100", "login_count = login_count + 1").where(id=1).sql()

# Q 条件
sql = SqlUtil.update("user").set(active=False).where(Q(status="banned") | Q(inactive_days__gt=365)).sql()

DELETE 链式删除¶

# 基础删除
sql = SqlUtil.delete("user").where(id=1).sql()

# 多条件
sql = SqlUtil.delete("user").where(status="inactive", created_at__lt="2020-01-01").sql()

# Q 条件
sql = SqlUtil.delete("user").where(Q(age__lt=18) | Q(status="banned")).sql()

# 原始 SQL
sql = SqlUtil.delete("user").where_raw("DATEDIFF(NOW(), created_at) > 365").sql()

CREATE TABLE 链式建表¶

使用 ColumnType 枚举自动适配不同方言的列类型。

from hutool import SqlUtil
from hutool.core.sql import ColumnType as CT

# MySQL
sql = (
    SqlUtil.create_table("user")
    .column("id", CT.SERIAL, primary_key=True)
    .column("name", CT.VARCHAR, type_args="(100)", nullable=False)
    .column("age", CT.INT, default=0)
    .column("bio", CT.TEXT)
    .column("active", CT.BOOLEAN, default=True)
    .column("extra", CT.JSON)
    .if_not_exists()
    .engine("InnoDB")
    .charset("utf8mb4")
    .sql()
)

# PostgreSQL
sql = (
    SqlUtil.create_table("user", dialect="postgresql")
    .column("id", CT.SERIAL, primary_key=True)
    .column("name", CT.VARCHAR, type_args="(100)", nullable=False)
    .column("uid", CT.UUID)
    .if_not_exists()
    .sql()
)

# 复合主键
sql = (
    SqlUtil.create_table("user_role")
    .column("user_id", CT.INT)
    .column("role_id", CT.INT)
    .primary_key("user_id", "role_id")
    .sql()
)

ColumnType 枚举值：

类型	MySQL	PostgreSQL	SQLite
`SERIAL`	`INT AUTO_INCREMENT`	`SERIAL`	`INTEGER`
`BIG_SERIAL`	`BIGINT AUTO_INCREMENT`	`BIGSERIAL`	`INTEGER`
`INT`	`INT`	`INTEGER`	`INTEGER`
`VARCHAR`	`VARCHAR`	`VARCHAR`	`TEXT`
`TEXT`	`TEXT`	`TEXT`	`TEXT`
`BOOLEAN`	`TINYINT(1)`	`BOOLEAN`	`INTEGER`
`JSON`	`JSON`	`JSONB`	`TEXT`
`JSONB`	`JSON`	`JSONB`	`TEXT`
`UUID`	`CHAR(36)`	`UUID`	`TEXT`
`DATETIME`	`DATETIME`	`TIMESTAMP`	`TEXT`
`BLOB`	`BLOB`	`BYTEA`	`BLOB`
`DECIMAL`	`DECIMAL`	`DECIMAL`	`NUMERIC`

备注

ColumnType.VARCHAR、ColumnType.CHAR、ColumnType.DECIMAL 等需要参数的类型，通过 .column(name, CT.VARCHAR, type_args='(255)') 传入参数。