`hutool.core.sql`¶

SQL 工具类。

提供 SQL 语句的生成工具，支持 SELECT、INSERT、UPDATE、DELETE 以及备份表创建。支持 MySQL、PostgreSQL、SQLite 三种方言。值格式化支持 Python 内置类型和可选的 numpy 类型（无需安装 numpy）。

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

Module Contents¶

Classes¶

`SqlUtil`	SQL 工具类，用于生成常用 SQL 语句。
`Q`	条件组合器，支持 `AND(&)`、`OR(\|)`、`NOT(~)`。
`F`	字段引用，支持算术运算和聚合函数。
`ColumnType`	列类型枚举，自动适配 MySQL / PostgreSQL / SQLite 方言。
`SqlBuilder`	SQL 链式构建器基类。
`SelectBuilder`	SELECT 链式构建器。
`InsertBuilder`	INSERT 链式构建器。
`UpdateBuilder`	UPDATE 链式构建器。
`DeleteBuilder`	DELETE 链式构建器。
`CreateTableBuilder`	CREATE TABLE 链式构建器。

Data¶

`_VALID_DIALECTS`
`_COLUMN_DIALECT_MAP`

API¶

hutool.core.sql._VALID_DIALECTS = ('mysql', 'postgresql', 'sqlite')¶

class hutool.core.sql.SqlUtil¶

SQL 工具类，用于生成常用 SQL 语句。

支持 MySQL、PostgreSQL、SQLite 三种方言。支持 INSERT ON DUPLICATE KEY UPDATE、REPLACE INTO、INSERT IGNORE 以及分页查询、备份表等常见操作。

static make_select_sql(table: str, columns: Union[List[str], str], condition: Optional[str] = None, order_by: Optional[list] = None, page_no: Optional[int] = None, page_size: Optional[int] = None, dialect: str = 'mysql') → str¶

生成查询 SQL。

参数:

table – 表名
columns – 要查询的字段名，列表或原始字符串
condition – WHERE 条件（不含 WHERE 关键字）
order_by – 排序字段列表
page_no – 页码（从 1 开始）
page_size – 每页记录数
dialect – SQL 方言，支持 "mysql"``（默认）、”postgresql”、”sqlite”``

返回:

查询 SQL 字符串

>>> SqlUtil.make_select_sql('user', ['name', 'age'])
'SELECT `name`,`age` FROM `user`'
>>> SqlUtil.make_select_sql('user', '*', page_no=2, page_size=10)
'SELECT * FROM `user` LIMIT 10, 10'
>>> SqlUtil.make_select_sql('user', '*', page_no=2, page_size=10, dialect='postgresql')
'SELECT * FROM "user" LIMIT 10 OFFSET 10'

static make_insert_sql(table: str, data: Union[dict, list], auto_update: bool = False, update_columns: tuple = (), update_columns_value: tuple = (), insert_ignore: bool = False, separator: str = ';\n', dialect: str = 'mysql', on_conflict: Optional[tuple] = None) → str¶

生成插入 SQL。

支持单条/批量插入、REPLACE INTO、INSERT IGNORE 以及 ON DUPLICATE KEY UPDATE。 PostgreSQL 和 SQLite 使用各自的 ON CONFLICT 语法。

参数:

table – 表名
data – 表数据，字典或字典列表
auto_update – 是否使用 REPLACE INTO（完全覆盖已有数据）
update_columns – ON DUPLICATE KEY UPDATE 时需要更新的列（指定后 auto_update 失效）
update_columns_value – 更新列的对应值，字符串需手动加引号
insert_ignore – 是否使用 INSERT IGNORE
separator – 多条 SQL 的分隔符
dialect – SQL 方言，支持 "mysql"``（默认）、”postgresql”、”sqlite”``
on_conflict – 冲突目标列元组，PostgreSQL 的 auto_update / update_columns 必须指定此参数

返回:

插入 SQL 字符串

>>> SqlUtil.make_insert_sql('user', {'name': 'Alice', 'age': 20})
"INSERT INTO `user` (`name`,`age`) VALUES ('Alice', 20);\n"
>>> SqlUtil.make_insert_sql('user', {'name': 'Alice'}, insert_ignore=True, dialect='postgresql')
'INSERT INTO "user" ("name") VALUES ('Alice') ON CONFLICT DO NOTHING;\n'

static make_update_sql(table: str, data: dict, condition: str, separator: str = ';\n', dialect: str = 'mysql') → str¶

生成更新 SQL。

参数:

table – 表名
data – 要更新的字段和值
condition – WHERE 条件（不含 WHERE 关键字）
separator – SQL 结尾分隔符
dialect – SQL 方言，支持 "mysql"``（默认）、”postgresql”、”sqlite”``

返回:

更新 SQL 字符串

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

static make_delete_sql(table: str, condition: str, dialect: str = 'mysql') → str¶

生成删除 SQL。

参数:

table – 表名
condition – WHERE 条件（不含 WHERE 关键字）
dialect – SQL 方言，支持 "mysql"``（默认）、”postgresql”、”sqlite”``

返回:

删除 SQL 字符串

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

static create_bak_table_sql(table: Union[str, List[str]], bak_table: Optional[Union[str, List[str]]] = None, bak_date: Optional[str] = None, separator: str = ';\n', dialect: str = 'mysql') → str¶

生成备份表 SQL（CREATE TABLE … AS SELECT）。

参数:

table – 原表名（支持字符串或列表）
bak_table – 备份表名，为空时自动使用 {table}_{bak_date} 格式
bak_date – 备份日期（格式 %Y%m%d），默认为当天
separator – 多条 SQL 的分隔符
dialect – SQL 方言，支持 "mysql"``（默认）、”postgresql”、”sqlite”``

返回:

创建备份表 SQL 字符串

抛出:

ValueError – 当 table 和 bak_table 长度不一致时

>>> SqlUtil.create_bak_table_sql('user', bak_date='20240101')
'create table user_20240101 as select * from user;\n'

static _validate_dialect(dialect: str) → None¶

校验方言参数是否合法。

参数:: dialect – 方言名称
抛出:: ValueError – 不支持的方言

>>> SqlUtil._validate_dialect('mysql')  # 正常
>>> SqlUtil._validate_dialect('oracle')  
Traceback (most recent call last):
...
ValueError: ...

static _quote(name: str, dialect: str = 'mysql') → str¶

根据方言引用标识符。

MySQL/SQLite 使用反引号，PostgreSQL 使用双引号。

参数:

name – 标识符名称
dialect – 方言

返回:

引用后的标识符

>>> SqlUtil._quote('user', 'mysql')
'`user`'
>>> SqlUtil._quote('user', 'postgresql')
'"user"'

static _conflict_cols(on_conflict: Optional[tuple], update_columns: tuple) → list¶

确定冲突目标列。

未指定 on_conflict 时使用 update_columns 作为冲突列。

参数:

on_conflict – 显式指定的冲突列
update_columns – 需要更新的列

返回:

冲突列列表

static _list2str(data: list) → str¶

将列表转为 SQL 元组字符串。

参数:: data – 数据列表
返回:: 元组格式字符串，如 (1, 2)

>>> SqlUtil._list2str([1, 2])
'(1, 2)'
>>> SqlUtil._list2str([1])
'(1)'

static _format_sql_value(value: Any) → Any¶

格式化 SQL 值。

处理以下类型： - str：包含子查询时用 ${} 包裹 - list / dict：转为 JSON 字符串 - datetime.date / datetime.time / datetime.datetime：格式化为字符串 - bool：转为 0 / 1 - numpy 浮点/整数类型（可选，需安装 numpy）

参数:: value – 原始值
返回:: 格式化后的值

>>> SqlUtil._format_sql_value(True)
1
>>> SqlUtil._format_sql_value({'key': 'value'})
'{"key": "value"}'

static select(table: str, dialect: str = 'mysql') → hutool.core.sql.SelectBuilder¶

创建 SELECT 链式构建器。

参数:

table – 表名
dialect – SQL 方言

返回:

SelectBuilder 实例

>>> sql = SqlUtil.select('user').columns('name', 'age').where(age__gt=18).sql()

static insert(table: str, dialect: str = 'mysql') → hutool.core.sql.InsertBuilder¶

创建 INSERT 链式构建器。

参数:

table – 表名
dialect – SQL 方言

返回:

InsertBuilder 实例

static update(table: str, dialect: str = 'mysql') → hutool.core.sql.UpdateBuilder¶

创建 UPDATE 链式构建器。

参数:

table – 表名
dialect – SQL 方言

返回:

UpdateBuilder 实例

static delete(table: str, dialect: str = 'mysql') → hutool.core.sql.DeleteBuilder¶

创建 DELETE 链式构建器。

参数:

table – 表名
dialect – SQL 方言

返回:

DeleteBuilder 实例

static create_table(table: str, dialect: str = 'mysql') → hutool.core.sql.CreateTableBuilder¶

创建 CREATE TABLE 链式构建器。

参数:

table – 表名
dialect – SQL 方言

返回:

CreateTableBuilder 实例

class hutool.core.sql.Q(_connector: str = 'AND', _negated: bool = False, _q_children: Optional[List[hutool.core.sql.Q]] = None, **kwargs)¶

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

参数:

_connector – 内部连接符 "AND" 或 "OR"
_negated – 是否取反
_q_children – 子 Q 对象列表（由 | / & 运算产生）
kwargs – 过滤条件，支持 field__op=value 语法

>>> (Q(age__gt=18) | Q(role='admin')).to_sql(SqlUtil._quote, SqlUtil._format_sql_value, 'mysql')
'(`age`>18 OR `role`='admin')'

Initialization

_OPS = None¶

__or__(other: hutool.core.sql.Q) → hutool.core.sql.Q¶

__and__(other: hutool.core.sql.Q) → hutool.core.sql.Q¶

__invert__() → hutool.core.sql.Q¶

to_sql(quote_fn, format_fn, dialect: str) → str¶

递归生成 SQL 条件字符串。

参数:

quote_fn – 标识符引用函数
format_fn – 值格式化函数
dialect – SQL 方言

返回:

SQL 条件片段

static _resolve_single_filter(key: str, value, quote_fn, format_fn, dialect: str) → str¶: 解析单个 field__op=value 条件。

class hutool.core.sql.F(name: str)¶

字段引用，支持算术运算和聚合函数。

参数:: name – 字段名

>>> (F('balance') + 100).to_sql(SqlUtil._quote, 'mysql')
'`balance`+100'
>>> F.count('*').to_sql(SqlUtil._quote, 'mysql')
'COUNT(*)'

Initialization

__add__(other)¶

__radd__(other)¶

__sub__(other)¶

__rsub__(other)¶

__mul__(other)¶

__rmul__(other)¶

__truediv__(other)¶

__rtruediv__(other)¶

__neg__()¶

_clone(op: str, value)¶

static count(name: str = '*') → hutool.core.sql.F¶: COUNT 聚合。

static sum(name: str) → hutool.core.sql.F¶: SUM 聚合。

static max(name: str) → hutool.core.sql.F¶: MAX 聚合。

static min(name: str) → hutool.core.sql.F¶: MIN 聚合。

static avg(name: str) → hutool.core.sql.F¶: AVG 聚合。

to_sql(quote_fn, dialect: str = 'mysql') → str¶

生成字段引用 SQL 片段。

参数:

quote_fn – 标识符引用函数
dialect – SQL 方言

返回:

SQL 表达式

class hutool.core.sql.ColumnType¶

Bases: enum.Enum

列类型枚举，自动适配 MySQL / PostgreSQL / SQLite 方言。

>>> ColumnType.VARCHAR.to_sql('mysql', '(100)')
'VARCHAR(100)'
>>> ColumnType.SERIAL.to_sql('postgresql')
'SERIAL'
>>> ColumnType.BOOLEAN.to_sql('mysql')
'TINYINT(1)'

TINYINT = 'TINYINT'¶

SMALLINT = 'SMALLINT'¶

MEDIUMINT = 'MEDIUMINT'¶

INT = 'INT'¶

BIGINT = 'BIGINT'¶

SERIAL = 'SERIAL'¶

BIG_SERIAL = 'BIG_SERIAL'¶

FLOAT = 'FLOAT'¶

DOUBLE = 'DOUBLE'¶

REAL = 'REAL'¶

DECIMAL = 'DECIMAL'¶

NUMERIC = 'NUMERIC'¶

CHAR = 'CHAR'¶

VARCHAR = 'VARCHAR'¶

TINYTEXT = 'TINYTEXT'¶

TEXT = 'TEXT'¶

MEDIUMTEXT = 'MEDIUMTEXT'¶

LONGTEXT = 'LONGTEXT'¶

TINYBLOB = 'TINYBLOB'¶

BLOB = 'BLOB'¶

MEDIUMBLOB = 'MEDIUMBLOB'¶

LONGBLOB = 'LONGBLOB'¶

BINARY = 'BINARY'¶

VARBINARY = 'VARBINARY'¶

BYTEA = 'BYTEA'¶

DATE = 'DATE'¶

TIME = 'TIME'¶

DATETIME = 'DATETIME'¶

TIMESTAMP = 'TIMESTAMP'¶

BOOLEAN = 'BOOLEAN'¶

JSON = 'JSON'¶

JSONB = 'JSONB'¶

UUID = 'UUID'¶

ENUM = 'ENUM'¶

to_sql(dialect: str, type_args: Optional[str] = None) → str¶

获取对应方言的 SQL 类型字符串。

参数:

dialect – SQL 方言
type_args – 类型参数，如 "(100)"、"(10,2)"

返回:

SQL 类型字符串

hutool.core.sql._COLUMN_DIALECT_MAP: Dict[hutool.core.sql.ColumnType, Dict[str, str]] = None¶

class hutool.core.sql.SqlBuilder(dialect: str = 'mysql')¶

SQL 链式构建器基类。

Initialization

_FILTER_OPS = None¶

_q(name: str) → str¶: 引用标识符。

_clone()¶: 浅拷贝当前状态。

_format_value(val: Any) → Any¶: 格式化值。

_resolve_conditions(conditions: list) → str¶: 将条件列表（Q 对象或原始字符串）转为 SQL WHERE 子句。

_resolve_single_filter(key: str, value: Any) → str¶: 解析单个 field__op=value 为 SQL 片段。

_resolve_column(col) → str¶: 解析列为 SQL 片段（F 对象或原始字符串或普通列名）。

abstract sql() → str¶: 生成 SQL 字符串。

class hutool.core.sql.SelectBuilder(table: str, dialect: str = 'mysql')¶

Bases: hutool.core.sql.SqlBuilder

SELECT 链式构建器。

>>> (SqlUtil.select('user')
...     .columns('id', 'name', 'age')
...     .where(age__gt=18)
...     .order_by('-age', 'name')
...     .limit(10)
...     .sql())
'SELECT `id`,`name`,`age` FROM `user` WHERE `age`>18 ORDER BY `age` DESC,`name` ASC LIMIT 10'

Initialization

columns(*cols) → hutool.core.sql.SelectBuilder¶

指定查询列。可传原始 SQL 表达式。

参数:: cols – 列名或 SQL 表达式
返回:: self（链式调用）

where(*args, **kwargs) → hutool.core.sql.SelectBuilder¶

添加 WHERE 条件。

参数:

args – Q 对象或原始 SQL 字符串
kwargs – field__op=value 格式的过滤条件

返回:

self（链式调用）

where_raw(condition: str) → hutool.core.sql.SelectBuilder¶: 添加原始 SQL WHERE 条件。

order_by(*fields) → hutool.core.sql.SelectBuilder¶

添加排序。前缀 - 表示 DESC。

参数:: fields – 排序字段，如 "-age", "name"
返回:: self（链式调用）

limit(n: int) → hutool.core.sql.SelectBuilder¶: 设置 LIMIT。

offset(n: int) → hutool.core.sql.SelectBuilder¶: 设置 OFFSET。

page(page_no: int, page_size: int) → hutool.core.sql.SelectBuilder¶: 按页码设置分页。page_no 从 1 开始。

distinct() → hutool.core.sql.SelectBuilder¶: SELECT DISTINCT。

group_by(*fields) → hutool.core.sql.SelectBuilder¶: 添加 GROUP BY。可传原始 SQL。

having(*args, **kwargs) → hutool.core.sql.SelectBuilder¶: 添加 HAVING 条件。参数同 where。

sql() → str¶: 生成 SELECT SQL。

class hutool.core.sql.InsertBuilder(table: str, dialect: str = 'mysql')¶

Bases: hutool.core.sql.SqlBuilder

INSERT 链式构建器。

>>> (SqlUtil.insert('user', dialect='postgresql')
...     .values(name='Alice', age=20)
...     .on_conflict('id')
...     .do_update('name', 'age')
...     .sql())
'INSERT INTO "user" ("name","age") VALUES ('Alice',20) ON CONFLICT (id) DO UPDATE SET "name"=EXCLUDED."name","age"=EXCLUDED."age"'

Initialization

values(**kwargs) → hutool.core.sql.InsertBuilder¶: 设置插入数据（单条）。

values_list(data: List[dict]) → hutool.core.sql.InsertBuilder¶: 设置批量插入数据。

on_conflict(*cols) → hutool.core.sql.InsertBuilder¶: 指定冲突目标列。

do_update(*cols, values: Optional[Tuple[str, ...]] = None) → hutool.core.sql.InsertBuilder¶

ON CONFLICT DO UPDATE SET 指定列。

参数:

cols – 需要更新的列
values – 更新列的对应值（可选，手动加引号的字符串）

do_nothing() → hutool.core.sql.InsertBuilder¶: ON CONFLICT DO NOTHING。

ignore() → hutool.core.sql.InsertBuilder¶: INSERT IGNORE / INSERT OR IGNORE。

replace() → hutool.core.sql.InsertBuilder¶: REPLACE INTO / INSERT OR REPLACE。

separator(sep: str) → hutool.core.sql.InsertBuilder¶: 设置 SQL 结尾分隔符。

sql() → str¶: 生成 INSERT SQL。

class hutool.core.sql.UpdateBuilder(table: str, dialect: str = 'mysql')¶

Bases: hutool.core.sql.SqlBuilder

UPDATE 链式构建器。

>>> (SqlUtil.update('user')
...     .set(name='Bob', age=30)
...     .where(id=1)
...     .sql())
"UPDATE `user` SET `name`='Bob',`age`=30 WHERE `id`=1;\n"

Initialization

set(**kwargs) → hutool.core.sql.UpdateBuilder¶

设置更新字段。值可以是 F 对象。

参数:: kwargs – field=value 格式
返回:: self（链式调用）

set_raw(*raw_sqls) → hutool.core.sql.UpdateBuilder¶: 设置原始 SQL SET 子句。

where(*args, **kwargs) → hutool.core.sql.UpdateBuilder¶: 添加 WHERE 条件。

where_raw(condition: str) → hutool.core.sql.UpdateBuilder¶: 添加原始 SQL WHERE 条件。

separator(sep: str) → hutool.core.sql.UpdateBuilder¶: 设置 SQL 结尾分隔符。

sql() → str¶: 生成 UPDATE SQL。

class hutool.core.sql.DeleteBuilder(table: str, dialect: str = 'mysql')¶

Bases: hutool.core.sql.SqlBuilder

DELETE 链式构建器。

>>> (SqlUtil.delete('user')
...     .where(status='inactive', created_at__lt='2020-01-01')
...     .sql())
"DELETE FROM `user` WHERE `status`='inactive' AND `created_at`<'2020-01-01'"

Initialization

where(*args, **kwargs) → hutool.core.sql.DeleteBuilder¶: 添加 WHERE 条件。

where_raw(condition: str) → hutool.core.sql.DeleteBuilder¶: 添加原始 SQL WHERE 条件。

sql() → str¶: 生成 DELETE SQL。

class hutool.core.sql.CreateTableBuilder(table: str, dialect: str = 'mysql')¶

Bases: hutool.core.sql.SqlBuilder

CREATE TABLE 链式构建器。

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

>>> from hutool.core.sql import ColumnType as CT
>>> (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)
...     .if_not_exists()
...     .sql())
'CREATE TABLE IF NOT EXISTS `user` (`id` INT AUTO_INCREMENT PRIMARY KEY,`name` VARCHAR(100) NOT NULL,`age` INT DEFAULT 0)'

Initialization

column(name: str, col_type: hutool.core.sql.ColumnType, type_args: Optional[str] = None, primary_key: bool = False, nullable: bool = True, default: Any = None, auto_increment: bool = False, unique: bool = False, comment: Optional[str] = None) → hutool.core.sql.CreateTableBuilder¶

定义列。

参数:

name – 列名
col_type – ColumnType 枚举
type_args – 类型参数，如 "(100)"、"(10,2)"
primary_key – 是否为主键
nullable – 是否允许 NULL
default – 默认值
auto_increment – 是否自增（MySQL 方言）
unique – 是否唯一
comment – 列注释（MySQL）

返回:

self（链式调用）

if_not_exists() → hutool.core.sql.CreateTableBuilder¶: 添加 IF NOT EXISTS。

primary_key(*cols) → hutool.core.sql.CreateTableBuilder¶: 设置复合主键。

engine(name: str) → hutool.core.sql.CreateTableBuilder¶: 设置存储引擎（MySQL）。

charset(name: str) → hutool.core.sql.CreateTableBuilder¶: 设置字符集（MySQL）。

separator(sep: str) → hutool.core.sql.CreateTableBuilder¶: 设置 SQL 结尾分隔符。

sql() → str¶: 生成 CREATE TABLE DDL。

hutool.core.sql¶

Module Contents¶

Classes¶

Data¶

API¶

`hutool.core.sql`¶