大家好,我是何三,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 + 自定义模板(灵活但维护成本高)
  • datasetrecords(简单易用的封装)

今天我们以 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”的方式,真的能让你事半功倍!