详解 pypika 模块：SQL 语句生成器，让你再也不用为拼接 SQL 语句而发愁
楔子
作为一名后端开发，日常工作中难免要和数据库打交道，而想要操作数据库，必然要构建 SQL 语句。你可以手动编写原生 SQL，也可以借助现有的第三方模块，比如 pypika。
本篇文章就来介绍一下 pypika 的用法，由于是第三方库，需要先安装，直接 pip install pypika 即可。
注：Python 还有一个第三库叫 pika，是专门用来连接 RabbitMQ 的，这两个名字虽然很像，但是之间没有任何关系。
简单的 SELECT 查询
构建 SELECT 语句的入口点是 pypika.Query，而查询数据的话必然要有两个关键信息：表和字段。
from pypika import Query
query = Query.from_("people").select("id", "name", "age")
print(query)
"""
SELECT "id","name","age" FROM "people"
"""
# 返回的是一个 QueryBuilder 对象
print(query.__class__)
"""
<class 'pypika.queries.QueryBuilder'>
"""
# 直接转成字符串即可
print(str(query))
"""
SELECT "id","name","age" FROM "people"
"""
# 或者调用 get_sql 方法
print(query.get_sql())
"""
SELECT "id","name","age" FROM "people"
"""
以上我们就构建了最简单的 SQL 语句，当然表和字段都是以字符串形式指定的，我们还可以使用对象。
from pypika import Query, Table, Field
table = Table("people")
fields = [Field("id"), Field("name"), Field("age")]
query = Query.from_(table).select(*fields)
print(query)
"""
SELECT "id","name","age" FROM "people"
"""
# 有了对象之后，我们就可以指定别名了
table = Table("data").as_("d")
fields = [Field("max_count").as_("max_cnt")]
query = Query.from_(table).select(*fields)
print(query)
"""
SELECT "max_count" "max_cnt" FROM "data" "d"
"""
当然目前只指定了表，如果希望在指定表的同时还指定数据库，要怎么做呢？
from pypika import Query, Table, Database
database = Database("fruits")
# 或者直接写成 Table("apple", "fruits") 也可以
table = Table("apple", database)
query = Query.from_(table).select("name", "price")
print(query)
"""
SELECT "name","price" FROM "fruits"."apple"
"""
# 注意：我们不要写成 Table("fruits.apple")，这是不对的
# 因为这表示从一张名为 "fruits.apple" 的表中获取数据
# 显然这是不符合规范的，表名不应该包含 . 这种特殊字符
table = Table("fruits.apple")
query = Query.from_(table).select("name", "price")
print(query)
"""
SELECT "name","price" FROM "fruits.apple"
"""
# 注意 from 子句，结果是 "fruits.apple"
# 真正的格式应该是 "fruits"."apple"，而不是 "fruits.apple"
如果是 MySQL，那么层级是数据库、表，但如果是 PostgreSQL，那么还会多一层 schema。如果想指定 schema，该怎么做呢？
from pypika import Query, Table, Database, Schema
# 数据库
database = Database("fruits")
# schema，隶属于指定的 database
schema = Schema("default", database)
# 表，隶属于指定的 schema
table = Table("apple", schema)
# 构建查询
query = Query.from_(table).select("name", "price")
print(query)
"""
SELECT "name","price" FROM "fruits"."default"."apple"
"""
然后在筛选字段的时候，还可以对字段做操作。
from pypika import Query, Field
# 如果想对字段做操作，那么需要使用 Field 对象
query = Query.from_("t").select(
Field("id") + 1,
Field("first") + Field("last"),
(Field("count") + 200) * Field("price")
)
print(query)
"""
SELECT "id"+1,"first"+"last",("count"+200)*"price" FROM "t"
"""
可以看到还是比较强大的，特别是会自动给你加上引号，这样可以防止关键字冲突。
WHERE 条件过滤
在获取数据的时候，很少会全量获取，绝大多数都是获取满足指定条件的数据，这个时候就需要使用 WHERE 语句。
from pypika import Query, Field
query = Query.from_("t").select("*").where(
(Field("salary") >= 10000) &
(Field("age").between(18, 30)) &
(Field("name").like("张%")) &
(Field("department").isin(["销售", "财务"]))
)
print(query)
"""
SELECT * FROM "t"
WHERE "salary">=10000 AND
"age" BETWEEN 18 AND 30 AND
"name" LIKE '张%' AND
"department" IN ('销售','财务')
"""
你在数据库中可以使用的语法，比如 IS NULL、NOT IN、IS NOT NULL 等等，在 Field 对象中都有指定的方法对应，并且这些方法都见名知意，可以自己试一下。
然后如果有多个条件，那么之间可以用 & 和 | 进行组合，等价于 AND 和 OR。
分组和聚合
接下来说一说 GROUP BY，既然提到它，那就必须要先了解如何在 pypika 中指定聚合函数。
from pypika import functions as fn, Field
fn.Count(Field("id"))
fn.Concat(Field("first_name"), "-", Field("last_name"))
fn.Substring(Field("name"), 3, 9)
所有的聚合函数，都可以在 functions 模块中找到。
from pypika import Field, Query
from pypika import functions as fn
query = Query.from_("people").select(
"age", fn.Count(Field("id"))
).where(
Field("age")[18: 30] & (Field("length") < 160)
).groupby("age")
print(query)
"""
SELECT "age",COUNT("id") FROM "people"
WHERE "age" BETWEEN 18 AND 30 AND "length"<160
GROUP BY "age"
"""
在指定字段的时候，可以直接传一个字符串，也可以传一个 Field 对象。有时候为了方便，当不需要对字段做操作的时候，我们会直接传一个字符串。
但对于 fn.Count 等聚合函数来说，里面一定要传 Field 对象，至于原因我们测试一下就知道了。
from pypika import Field, Query
from pypika import functions as fn
query1 = Query.from_("t").select(fn.Substring("name", 1, 5))
query2 = Query.from_("t").select(fn.Substring(Field("name"), 1, 5))
# SUBSTRING 里面的第一个参数表示长度为 4 的字符串
print(query1)
"""
SELECT SUBSTRING('name',1,5) FROM "t"
"""
# SUBSTRING 里面的第一个参数表示字段 name
print(query2)
"""
SELECT SUBSTRING("name",1,5) FROM "t"
"""
这就是两者的区别，当然有人会觉得这是 PostgreSQL 的语法吧，MySQL 应该是反引号才对，没错，后面的话我们会说如何适配数据库。因为数据库的种类不同，语法也会稍有不同，而目前没有任何信息表明我们使用的到底是哪一种数据库。
当执行了 GROUP BY 之后，还可以继续执行 HAVING。
from pypika import Field, Query
from pypika import functions as fn
query = Query.from_("people").select(
"age", fn.Count(Field("id"))
).groupby("age").having(fn.Count(Field("id")) > 30)
print(query)
"""
SELECT "age",COUNT("id") FROM "people"
GROUP BY "age" HAVING COUNT("id")>30
"""
以上就是分组和聚合。
两表 JOIN
如果是两张表需要 JOIN 的话，该怎么做呢？
from pypika import Query, Table
t1 = Table("t1")
t2 = Table("t2")
query = Query.from_(t1).select(
t1.name, t2.age
).left_join(t2).using("id")
print(query)
"""
SELECT "t1"."name","t2"."age" FROM "t1"
LEFT JOIN "t2" USING ("id")
"""
由于涉及到多张表，那么当字段出现重叠的时候，需要同时指定表名，可以直接通过获取 Table 对象属性的方式指定。但如果表的字段名恰好和 Table 对象的某个属性名冲突，就不行了，我们举个例子。
from pypika import Query, Table
t1 = Table("t1")
t2 = Table("t2")
# 比如 Table 对象有一个属性叫 field
# 而表中也有一个字段叫 field
query = Query.from_(t1).select(
t2.field
).left_join(t2).using("id")
print(query)
"""
SELECT <bound method Selectable.field of Table('t2')>
FROM "t1" LEFT JOIN "t2" USING ("id")
"""
print(t2.field)
"""
<bound method Selectable.field of Table('t2')>
"""
显然这个时候就比较尴尬了，那我们应该怎么做呢？
from pypika import Query, Table, Field
t1 = Table("t1")
t2 = Table("t2")
query = Query.from_(t1).select(
Field("field", table=t2)
).left_join(t2).using("id")
print(query)
"""
SELECT "t2"."field" FROM "t1" LEFT JOIN "t2" USING ("id")
"""
这样就没问题了，Field 类还可以接收一个 table 参数，指定字段来自于哪张表，当然如果是单表，那么该字段就无需指定了。
当然除了 LEFT JOIN 之外，其它 JOIN 也是支持的。
这些方法内部都调用了 join 方法。
如果两张表要连接的字段的名字相同、并且是等值连接，那么可以使用 using。但还有一种情况是：两个名字不同的字段进行等值连接，比如一张表的 uid 等于另一张表的 tid 等等。
from pypika import Query, Table
t1 = Table("t1")
t2 = Table("t2")
# Field("uid", table=t1) 还可以写成 t1.field("uid")
# 这两者是完全等价的，但 t1.field("uid") 写起来更方便
query = Query.from_(t1).select(t2.age, t1.name) \
.left_join(t2) \
.on(t1.field("uid") == t2.field("tid")) \
.where(t1.age > 18)
print(query)
"""
SELECT "t2"."age","t1"."name" FROM "t1"
LEFT JOIN "t2" ON "t1"."uid"="t2"."tid"
WHERE "t1"."age">18
"""
以上就是 JOIN 相关的内容，至于三表连接，你可以自己试一下。
嵌套子查询
再来看看嵌套子查询：
from pypika import Query, Table, functions as fn
t1 = Table("t1")
t2 = Table("t2")
sub_query = Query.from_(t1).select(fn.Avg(t2.age).as_("avg")) \
.left_join(t2).using("id").where(t1.age > 18)
print(sub_query)
"""
SELECT AVG("t2"."age") "avg" FROM "t1"
LEFT JOIN "t2" USING ("id") WHERE "t1"."age">18
"""
# 子查询完全可以当成一张表来操作
query = Query.from_(t1).select("age", "name").where(
t1.field("age") > Query.from_(sub_query).select("avg")
)
print(query)
"""
SELECT "age","name" FROM "t1"
WHERE "age">(
SELECT "sq0"."avg" FROM (
SELECT AVG("t2"."age") "avg"
FROM "t1" LEFT JOIN "t2" USING ("id") WHERE "t1"."age">18
) "sq0"
)
"""
集合运算
两个结果集之间是可以合并的，比如 UNION 和 UNION ALL，至于 UNION DISTINCE 是 UNION 的同义词，所以 pypika 没有设置专门的函数。另外 UNION 虽然可以用来合并多个结果集，但前提是它们要有相同的列。
from pypika import Query, Table
t1 = Table("t1")
t2 = Table("t2")
query1 = Query.from_(t1).select("name", "salary")
query2 = Query.from_(t2).select("name", "salary")
print(query1.union(query2))
print(query2.union(query1))
"""
(SELECT "name","salary" FROM "t1") UNION (SELECT "name","salary" FROM "t2")
(SELECT "name","salary" FROM "t2") UNION (SELECT "name","salary" FROM "t1")
"""
# union 可以使用 + 代替
print(str(query1 + query2) == str(query1.union(query2))) # True
print(str(query2 + query1) == str(query2.union(query1))) # True
# union_all 可以使用 * 代替
print(query1.union_all(query2))
print(query2.union_all(query1))
"""
(SELECT "name","salary" FROM "t1") UNION ALL (SELECT "name","salary" FROM "t2")
(SELECT "name","salary" FROM "t2") UNION ALL (SELECT "name","salary" FROM "t1")
"""
print(str(query1 * query2) == str(query1.union_all(query2))) # True
print(str(query2 * query1) == str(query2.union_all(query1))) # True
此外还有交集、差集、对称差集。
from pypika import Query, Table
t1 = Table("t1")
t2 = Table("t2")
query1 = Query.from_(t1).select("name", "salary")
query2 = Query.from_(t2).select("name", "salary")
# 交集，没有提供专门的操作符
print(query1.intersect(query2))
"""
(SELECT "name","salary" FROM "t1")
INTERSECT
(SELECT "name","salary" FROM "t2")
"""
# 差集，可以使用减号替代
print(query1.minus(query2))
"""
(SELECT "name","salary" FROM "t1")
MINUS
(SELECT "name","salary" FROM "t2")
"""
# 对称差集，没有提供专门的操作符
print(query1.except_of(query2))
"""
(SELECT "name","salary" FROM "t1")
EXCEPT
(SELECT "name","salary" FROM "t2")
"""
时间间隔
有时我们查找数据需要查找 7 天以内的，或者 1 个月以内的，这时候该怎么做呢？
from pypika import Query, Table, functions as fn, Interval
t = Table('fruits')
query = Query.from_(t).select(t.id, t.name) \
.where(t.harvest_date + Interval(months=1) < fn.Now())
print(query)
"""
SELECT "id","name" FROM "fruits"
WHERE "harvest_date"+INTERVAL '1 MONTH'<NOW()
"""
多值比较
SQL 有一个非常有用的特性，假设一张表中有 year、month 这两个字段，然后我想找出 year、month 组合起来之后大于 2020 年 7 月的记录。比如 year = 2021、month = 2 这条记录就是合法的，因为 year 是大于 2020 的。而 year = 2020、month = 8 也是合法的。
显然这个时候就有些不好搞了，我们无法通过 year > 2020 and month > 7 这种形式，但是数据库提供了多值比较：
select * from t where (year, month) > (2020, 7)
是不是很像元组呢？会先比较 year，如果满足 year > 2020，直接成立。year < 2020，直接不成立，后面就不用比了。如果 year = 2020，那么再比较 month。
from pypika import Query, Table, Tuple
t = Table("t")
query = Query.from_(t).select(t.salary).where(
Tuple(t.year, t.month) >= (2020, 7))
print(query)
"""
SELECT "salary" FROM "t" WHERE ("year","month")>=(2020,7)
"""
对于 in 字句也是同样的道理：
from pypika import Query, Table, Tuple
t = Table("t")
query = Query.from_(t).select(t.salary).where(
Tuple(t.year, t.month
).isin([(2020, 7), (2020, 8), (2020, 9)]))
print(query)
"""
SELECT "salary" FROM "t"
WHERE ("year","month") IN ((2020,7),(2020,8),(2020,9))
"""
CASE WHEN
然后看看 CASE WHEN，SQL 层面上的就不说了，我们只看怎么用 pypika 实现。
from pypika import Table, Query, Case
t = Table("t")
query = Query.from_(t).select(
t.name,
Case().when(t.age < 18, "未成年").when(t.age < 30, "成年")
.when(t.age < 50, "中年").else_("老年").as_("age")
)
print(query)
"""
SELECT "name",
CASE WHEN "age"<18 THEN '未成年'
WHEN "age"<30 THEN '成年'
WHEN "age"<50 THEN '中年'
ELSE '老年' END "age"
FROM "t"
"""
WITH 语句
WITH 语句就是给子查询指定一个名字，然后在其它地方可以直接使用该名字，就像访问一张已存在的表一样。
from pypika import Table, Query, AliasedQuery
t = Table("t")
sub_query = Query.from_(t).select("*")
query = Query.with_(sub_query, "alias").from_(
AliasedQuery("alias")).select("*")
print(query)
"""
WITH alias AS (SELECT * FROM "t") SELECT * FROM alias
"""
DISTINCT
如果我们想对结果集进行去重的话，要怎么做呢？
from pypika import Query, Table
t = Table("t")
# 只需要在 select 之前调用一次 distinct 即可
query = Query.from_(t).distinct().select(t.id, t.age)
print(query)
"""
SELECT DISTINCT "id","age" FROM "t"
"""
ORDER BY 排序
在查询到结果集之后，也可以进行排序。
from pypika import Query, Order
query = Query.from_("t").select("id", "name") \
.orderby("id", order=Order.desc)
print(query)
"""
SELECT "id","name" FROM "t" ORDER BY "id" DESC
"""
# 如果是多个字段的话
query = Query.from_("t").select("id", "name") \
.orderby("age", "id")
print(query)
"""
SELECT "id","name" FROM "t" ORDER BY "age","id"
"""
query = Query.from_("t").select("id", "name") \
.orderby("age", "id", order=Order.desc)
print(query)
"""
SELECT "id","name" FROM "t" ORDER BY "age" DESC,"id" DESC
"""
# 如果是一个字段升序、一个字段降序怎么办？很简单，调用两次 orderby 即可
query = Query.from_("t").select("id", "name") \
.orderby("age", order=Order.desc).orderby("id")
print(query)
"""
SELECT "id","name" FROM "t" ORDER BY "age" DESC,"id"
"""
LIMIT 和 OFFSET
获取到结果集之后，可以选择指定的条数，比如实现分页功能。
from pypika import Table, Query, Field
from pypika import functions as fn, Order
table = Table("t")
query = Query.from_(table) \
.select(fn.Count(Field("id")).as_("count"), "age", "length") \
.where(table.field("age") > 18) \
.groupby("age", "length") \
.having(fn.Count("id") > 10) \
.orderby("count", order=Order.desc) \
.orderby("age", order=Order.asc) \
.limit(10).offset(5)
print(query)
"""
SELECT COUNT("id") "count","age","length"
FROM "t" WHERE "age">18
GROUP BY "age","length"
HAVING COUNT('id')>10
ORDER BY "count" DESC,"age" ASC
LIMIT 10 OFFSET 5
"""
这里我们将所有子句都演示了一遍，算是做一个总结。
插入数据
以上说的都是查询数据，那么插入数据要怎么实现呢？
from pypika import Table, Query
t = Table("t")
# 查询是 Query.from_，插入数据是 Query.into
query = Query.into(t).insert(1, "古明地觉", 16, "东方地灵殿")
print(query)
"""
INSERT INTO "t" VALUES (1,'古明地觉',16,'东方地灵殿')
"""
# 如果存在 None 值，会自动处理
query = Query.into(t).insert(1, "古明地觉", None, "东方地灵殿")
print(query)
"""
INSERT INTO "t" VALUES (1,'古明地觉',NULL,'东方地灵殿')
"""
如果表中存在 JSON，那么直接对字典 dumps 一下传进去即可。
当然上面是单条插入，如果我想同时插入多条数据，该怎么做呢？
from pypika import Table, Query
table = Table("t")
query = Query.into(table) \
.insert(1, "古明地觉", 16, "东方地灵殿") \
.insert(2, "古明地恋", 15, "东方地灵殿")
print(query)
"""
INSERT INTO "t"
VALUES (1,'古明地觉',16,'东方地灵殿'),
(2,'古明地恋',15,'东方地灵殿')
"""
# 或者
query = Query.into(table).insert((1, "古明地觉", 16, "东方地灵殿"),
(2, "古明地恋", 15, "东方地灵殿"))
print(query)
"""
INSERT INTO "t"
VALUES (1,'古明地觉',16,'东方地灵殿'),
(2,'古明地恋',15,'东方地灵殿')
"""
在插入数据的时候，也可以选择指定的部分字段。
from pypika import Table, Query, Field
table = Table("t")
query = Query.into(table).columns(
"id", table.field("name"), table.age, Field("place")
).insert(1, "古明地觉", 16, "东方地灵殿")
print(query)
"""
INSERT INTO "t" ("id","name","age","place")
VALUES (1,'古明地觉',16,'东方地灵殿')
"""
当然也可以将一张表的记录插入到另一张表中。
from pypika import Table, Query, Field
t1 = Table("t1")
t2 = Table("t2")
query = Query.into(t1).columns("id", "name", "age") \
.from_(t2).select("id", "name", "age") \
.where(Field("age") > 18)
print(query)
"""
INSERT INTO "t1" ("id","name","age")
SELECT "id","name","age" FROM "t2" WHERE "age">18
"""
两个表 JOIN 之后的结果也可以插入到新表中，不过在 Python 中拼接 SQL 语句的时候，很少会遇到这种需求。
更新数据
再来看看更新数据怎么做？
from pypika import Table, Query
t = Table("t")
# 更新是 update
query = Query.update(t).set(t.name, "古明地恋")
print(query)
"""
UPDATE "t" SET "name"='古明地恋'
"""
query = Query.update(t).set(t.name, "古明地恋").where(t.id == 1)
print(query)
"""
UPDATE "t" SET "name"='古明地恋' WHERE "id"=1
"""
query = Query.update(t).set(t.name, "古明地恋").set(t.age, 16)
print(query)
"""
UPDATE "t" SET "name"='古明地恋',"age"=16
"""
用另一张表的数据更新当前也是一种比较常见的操作，比如 t1 有 uid、name 两个字段，t2 有 tid、name 两个字段。如果 t1 的 uid 在 t2 的 tid 中存在，那么就用 t2 的 name 更新掉 t1 的 name。
from pypika import Table, Query
t1 = Table("t1")
t2 = Table("t2")
query = Query.update(t1).join(t2).on(
t1.uid == t2.tid
).set(t1.name, t2.name).where(t1.uid > 10)
print(query)
"""
UPDATE "t1" JOIN "t2" ON "t1"."uid"="t2"."tid"
SET "name"="t2"."name" WHERE "t1"."uid">10
"""
数据库适配
不同数据库的 SQL 语法会有略微不同，最大的一个不同就是包裹字段所用的符号，MySQL 用的是反引号、PostgreSQL 用的是双引号。而 pypika 不知道你的数据库种类，所以默认用的是双引号。如果想适配 MySQL 的话，那么应该告诉 pypika，我们要适配 MySQL。
from pypika import (
MySQLQuery,
PostgreSQLQuery,
OracleQuery,
MSSQLQuery,
SQLLiteQuery,
ClickHouseQuery,
VerticaQuery
)
# pypika 提供多种数据库的适配，我们以 MySQL 为例
# 之前用的是 Query 这个类，而以上这些类都继承 Query
# 所以语法和之前是一样的
from pypika import Table, MySQLQuery, PostgreSQLQuery
t = Table("t")
print(
MySQLQuery.from_(t).select(t.id, t.age)
) # SELECT `id`,`age` FROM `t`
print(
PostgreSQLQuery.from_(t).select(t.id, t.age)
) # SELECT "id","age" FROM "t"
要操作哪一种数据库，直接选择对应的 Query 即可。
小结
以上就是 pypika 的相关内容，总的来说还是很方便的，在面对一些不复杂的 SQL 时，使用该模块会非常方便。
当然 pypika 还支持更多高级用法，比如窗口函数，有兴趣可以查看官网。
相关推荐
