詳解 pypika 模塊：SQL 語句生成器，讓你再也不用為拼接 SQL 語句而發愁

作者：古明地覺 2024-07-09 08:49:56

不同數據庫的 SQL 語法會有略微不同，最大的一個不同就是包裹字段所用的符號，MySQL 用的是反引號、PostgreSQL 用的是雙引號。而 pypika 不知道你的數據庫種類，所以默認用的是雙引號。如果想適配 MySQL 的話，那么應該告訴 pypika，我們要適配 MySQL。

楔子

作為一名后端開發，日常工作中難免要和數據庫打交道，而想要操作數據庫，必然要構建 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(mnotallow=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
"""

數據庫適配

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 還支持更多高級用法，比如窗口函數，有興趣可以查看官網。

https://pypika.readthedocs.io

責任編輯：武曉燕來源：古明地覺的編程教室

語句生成器 SQL

成人免费xxxxx在线视频软件_久久精品久久久_亚洲国产精品久久久_天天色天天色_亚洲人成一区_欧美一级欧美三级在线观看