大家好,我是何三,80后老猿,独立开发者
在日常开发中,操作数据库是每个程序员都绕不开的任务。无论是增删改查(CRUD),还是复杂的数据分析查询,我们几乎每天都要跟 SQL 打交道。
但你有没有发现一个问题?SQL 写多了其实很重复、容易出错,而且调试起来也挺麻烦的。
今天,我想分享一个我最近常用的技巧:用 Python 自动生成 SQL 语句。不仅提升了开发效率,还减少了手动出错的概率。
一、为什么我们要“自动”生成 SQL?
很多人可能会问:“SQL 不就是写写SELECT FROM WHERE 吗?有必要自动化吗?”
确实,对于简单的查询来说,手写 SQL 没什么问题。但在以下几种场景下,手写 SQL 就显得力不从心了:
- 需要根据条件动态拼接 SQL(比如搜索页面)
- 表结构频繁变更,SQL 也要跟着改
- 多个模块需要复用相似逻辑
- 需要兼容多种数据库(MySQL / PostgreSQL / SQLite)
这时候,自动生成 SQL 的能力就变得非常有价值了。
二、Python 怎么帮我们生成 SQL?
Python 有很多库可以帮助我们生成 SQL,最常用的是:
SQLAlchemy
(ORM 工具,支持 SQL 生成)sqlglot
(轻量级 SQL 解析与生成工具)jinja2
+ 自定义模板(灵活但维护成本高)dataset
或records
(简单易用的封装)
今天我们以 sqlglot
为例,来展示如何用 Python 动态生成 SQL。
三、什么是 sqlglot?
📘 官方地址:https://github.com/tobymao/sqlglot
sqlglot
是一个轻量级、快速且功能强大的 SQL 解析器和生成器,支持多种数据库语法(如 MySQL、PostgreSQL、BigQuery 等),并且具有良好的可扩展性。
它不仅能将字符串解析为 AST(抽象语法树),还能将 AST 转换为不同数据库的 SQL 语句,实现跨平台兼容。
✅ 主要特性:
特性 | 描述 |
---|---|
SQL 解析 | 支持多种数据库方言 |
SQL 生成 | 可构建复杂的 SQL 查询 |
数据库转换 | 可将一种数据库的 SQL 转换为另一种 |
高性能 | 极快的解析速度 |
易于扩展 | 可自定义函数、关键字等 |
四、实战演示:用 sqlglot 自动生成 SQL 查询
安装 sqlglot
pip install sqlglot
示例1:基本查询生成
我们先来看一个最简单的例子,用 Python 构建一个 SELECT 查询:
import sqlglot
from sqlglot import exp, parse_one
# 构造 SELECT * FROM users WHERE age > 30
query = (
exp.select("*")
.from_("users")
.where(exp.column("age") > 30)
)
print(query.sql())
输出结果:
SELECT * FROM users WHERE age > 30
是不是很简洁?你可以通过链式调用的方式构建复杂的 SQL。
示例2:动态拼接查询条件
在实际项目中,经常会有多个筛选条件,用户可能只填了一部分。
我们可以用 sqlglot
动态添加 where 条件:
from sqlglot import exp
conditions = []
if age_min := 25:
conditions.append(exp.column("age") >= age_min)
if name_like := "John":
conditions.append(exp.column("name").like(f"%{name_like}%"))
query = exp.select("*").from_("users")
if conditions:
query = query.where(*conditions)
print(query.sql())
输出:
SELECT * FROM users WHERE age >= 25 AND name LIKE '%John%'
这样就可以实现一个可扩展、可复用的查询构造器,避免手动拼接字符串带来的安全风险(如 SQL 注入)。
示例3:聚合函数 & 分组统计
query = (
exp.select("department", exp.func("AVG", "salary").as_("avg_salary"))
.from_("employees")
.group_by("department")
.having(exp.func("COUNT", "*") > 5)
)
print(query.sql())
输出:
SELECT department, AVG(salary) AS avg_salary FROM employees GROUP BY department HAVING COUNT(*) > 5
示例4:JOIN 操作
query = (
exp.select("u.id", "u.name", "o.order_id")
.from_("users u")
.join("orders o", on=exp.column("u.id", table="u") == exp.column("user_id", table="o"))
)
print(query.sql())
输出:
SELECT u.id, u.name, o.order_id FROM users AS u JOIN orders AS o ON u.id = o.user_id
示例5:INSERT 语句构造
query = (
exp.insert(
exp.values([("Alice", 30), ("Bob", 28)], columns=["name", "age"]),
into="users"
)
)
print(query.sql())
输出:
INSERT INTO users (name, age) VALUES ('Alice', 30), ('Bob', 28)
示例6:跨数据库兼容性处理
sqlglot
还支持不同数据库之间的语法转换,比如将 MySQL 转换为 Postgres:
expression = sqlglot.parse_one("SELECT a FROM test WHERE a IN (SELECT a FROM b LIMIT 10)")
# 转成 Postgres 语法
pg_sql = expression.sql(dialect="postgres")
print(pg_sql)
输出:
SELECT a FROM test WHERE a IN (SELECT a FROM b LIMIT 10)
虽然在这个例子里没有变化,但在某些函数名或关键字上,不同的数据库有差异。使用 sqlglot
可以帮助我们在不同数据库之间保持一致的代码逻辑。
五、进阶技巧:封装成通用查询构造器
我们可以进一步封装上面的功能,写一个通用的查询构造类:
class QueryBuilder:
def __init__(self, table):
self.table = table
self._select = exp.select("*").from_(table)
self.conditions = []
def add_condition(self, condition):
self.conditions.append(condition)
return self
def build(self):
if self.conditions:
self._select = self._select.where(*self.conditions)
return self._select.sql()
# 使用示例
builder = QueryBuilder("users")
builder.add_condition(exp.column("age") > 25)
builder.add_condition(exp.column("status") == "active")
print(builder.build())
输出:
SELECT * FROM users WHERE age > 25 AND status = 'active'
这样一来,我们可以根据不同业务需求快速生成对应的 SQL,而不需要每次都手动拼接。
最后
如果你也在做数据平台、报表系统、后台管理系统,或者经常需要写大量 SQL,不妨试试这种“用 Python 自动生成 SQL”的方式,真的能让你事半功倍!