靈活使用 SQLAlchemy 中的 ORM 查詢

之前做查詢一直覺得直接拼 SQL 比較方便，用了 SQLAlchemy 的 ORM 查詢之後，發現也還可以，還提高了可讀性。

這篇文章主要說說 SQLAlchemy 常用的 ORM 查詢方式，偏實踐。看了之後，對付開發中的查詢需求，我覺得可以滿足不少。

為方便說明，假設有如下數據

圖書表 books

+----+--------+--------------------------+-------+| id | cat_id | name | price |+----+--------+--------------------------+-------+| 1 | 1 | 生死疲勞 | 40.40 || 2 | 1 | 皮囊 | 31.80 || 3 | 2 | 半小時漫畫中國史 | 33.60 || 4 | 2 | 耶路撒冷三千年 | 55.60 || 5 | 2 | 國家寶藏 | 52.80 || 6 | 3 | 時間簡史 | 31.10 || 7 | 3 | 宇宙簡史 | 22.10 || 8 | 3 | 自然史 | 26.10 || 9 | 3 | 人類簡史 | 40.80 || 10 | 3 | 萬物簡史 | 33.20 |+----+--------+--------------------------+-------+

分類表 categories

+----+--------------+| id | name |+----+--------------+| 1 | 文學 || 2 | 人文社科 || 3 | 科技 |+----+--------------+

ORM 對象定義如下

---

注意：本文 Python 代碼在以下環境測試通過

• Python 3.6.0
• PyMySQL 0.8.1
• SQLAlchemy 1.2.8

---

39;mysql+pymysql://username:password&39;@127.0.0.1:3306/db_name?charset=utf8&39;books&39;name&39;price&39;categories&39;name&39;id&39;cat_id&39;name&39;生死疲勞&39;price&39;40.40&39;id&39;cat_id&39;name&39;生死疲勞&39;price&39;40.40&39;id&39;cat_id&39;name&39;皮囊&39;price&39;31.80& 找不到記錄會拋如下錯誤 找到多條記錄會拋如下錯誤 正常，得到如下結果39;id&39;cat_id&39;name&39;萬物簡史& &39;: Decimal(&39;)}book = session \ .query(Book).filter(Book.id == 10) \ .one()print(book and book.to_dict())

• count() 返回記錄條數

count = session \ .query(Book) \ .filter(Book.cat_id == 3) \ .count()print(count)

結果

5

• limit() 限制返回的記錄條數

books = session \ .query(Book) \ .filter(Book.cat_id == 3) \ .limit(3) \ .all()print([v.to_dict() for v in books])

結果

[{&39;: 6, &39;: 3, &39;: &39;, &39;: Decimal(&39;)}, {&39;: 7, &39;: 3, &39;: &39;, &39;: Decimal(&39;)}, {&39;: 8, &39;: 3, &39;: &39;, &39;: Decimal(&39;)}]

• distinct() 與 SQL 的 distinct 語句行為一致

books = session \ .query(Book.cat_id) \ .distinct(Book.cat_id) \ .all()print([dict(zip(v.keys(), v)) for v in books])

結果

[{&39;: 1}, {&39;: 2}, {&39;: 3}]

• order_by() 將記錄按照某個欄位進行排序

如果要升序排列，去掉 .desc() 即可books = session \ .query(Book.id, Book.name) \ .filter(Book.id > 8) \ .order_by(Book.id.desc()) \ .all()print([dict(zip(v.keys(), v)) for v in books])

結果

[{&39;: 10, &39;: &39;}, {&39;: 9, &39;: &39;}]

• scalar() 返回調用 one() 後得到的結果的第一列值

book_name = session \ .query(Book.name) \ .filter(Book.id == 10)\ .scalar()print(book_name)

結果

萬物簡史

• exist() 查看記錄是否存在

39;id&39;cat_id&39;name&39;生死疲勞&39;price&39;40.40&39;id&39;cat_id&39;name&39;皮囊&39;price&39;31.80&39;id&39;cat_id&39;name&39;耶路撒冷三千年&39;price&39;55.60&39;id&39;cat_id&39;name&39;國家寶藏&39;price&39;52.80&39;id&39;cat_id&39;name&39;人類簡史&39;price&39;40.80&39;id&39;cat_id&39;name&39;耶路撒冷三千年&39;price&39;55.60&39;id&39;cat_id&39;name&39;宇宙簡史&39;price&39;22.10& 請求參數，這裡只是佔位，實際由用戶提交的請求決定params = {&39;: 1}conditions = []if params.get(&39;, 0): conditions.append(Book.cat_id == params[&39;])if params.get(&39;, 0): conditions.append(Book.price == params[&39;])if params.get(&39;, 0): conditions.append(Book.price >= params[&39;])if params.get(&39;, 0): conditions.append(Book.price <= params[&39;])books = session.query(Book).filter(*conditions).all()print([v.to_dict() for v in books])

結果

[{&39;: 1, &39;: 1, &39;: &39;, &39;: Decimal(&39;)}, {&39;: 2, &39;: 1, &39;: &39;, &39;: Decimal(&39;)}]

OR 查詢類似，將列表解包傳給 or_() 即可。

如果需求更複雜，AND 和 OR 都可能出現，這個時候根據情況多建幾個列表實現。這裡只向大家說明大致的思路，就不舉具體的例子了。

當然，如果都是等值查詢的話，比如只有這兩種情況

• 如果接收到非 0 的 cat_id，需要限制 cat_id 等於 0
• 如果接收到非 0 的 price，需要限制 price 等於傳入的 price

可以使用字典的解包給 filter_by() 傳參

39;price&39;cat_id&39;cat_id&39;cat_id&39;price&39;price&39;price&39;id&39;cat_id&39;name&39;時間簡史&39;price&39;31.10& 查詢 ID 在 1、3、5 中的記錄books = session.query(Book) \ .filter(Book.id.in_([1, 3, 5])) \ .all()

• INSTR()

39;時間簡史& 查詢名稱包含「時間簡史」的圖書 FIND_IN_SET() 一般用於逗號分隔的 ID 串查找39;時間簡史& 查詢名稱以「簡史」結尾的圖書books = session.query(Book) \ .filter(Book.name.like(&39;)) \ .all()

• NOT

上面的 IN、INSTR、FIN_IN_SET、LIKE 都可以使用 ~ 符號取反。比如

39;簡史&39;id&39;name&39;時間簡史&39;id&39;name&39;宇宙簡史&39;id&39;name&39;人類簡史&39;id&39;name&39;萬物簡史& 如果 ORM 對象中定義有外鍵關係 否則，必須要 books = session \ .query(Book.id, Book.name.label(&39;), Category.name.label(&39;)) \ .join(Category, Book.cat_id == Category.id) \ .filter(Category.name == &39;, Book.price > 40) \ .all()print([dict(zip(v.keys(), v)) for v in books])

結果

[{&39;: 9, &39;: &39;, &39;: &39;}]

統計各個分類的圖書的數量

from sqlalchemy import funcbooks = session \ .query(Category.name.label(&39;), func.count(Book.id).label(&39;)) \ .join(Book, Category.id == Book.cat_id) \ .group_by(Category.id) \ .all()print([dict(zip(v.keys(), v)) for v in books])

結果

[{&39;: &39;, &39;: 2}, {&39;: &39;, &39;: 3}, {&39;: &39;, &39;: 5}]

9.2 外連接

為方便說明，我們僅在這一小節中向 books 表中加入如下數據

+----+--------+-----------------+-------+| id | cat_id | name | price |+----+--------+-----------------+-------+| 11 | 5 | 人性的弱點 | 54.40 |+----+--------+-----------------+-------+

查看 ID 大於等於 9 的圖書的分類信息

如果 ORM 對象中定義有外鍵關係 否則，必須要books = session \ .query(Book.id.label(&39;), Book.name.label(&39;), Category.id.label(&39;), Category.name.label(&39;)) \ .outerjoin(Category, Book.cat_id == Category.id) \ .filter(Book.id >= 9) \ .all()print([dict(zip(v.keys(), v)) for v in books])

結果

[{&39;: 9, &39;: &39;, &39;: 3, &39;: &39;}, {&39;: 10, &39;: &39;, &39;: 3, &39;: &39;}, {&39;: 11, &39;: &39;, &39;: None, &39;: None}]

注意最後一條記錄。

10 列印 SQL

當碰到複雜的查詢，比如有 AND、有 OR、還有連接查詢時，有時可能得不到預期的結果，這時我們可以打出最終的 SQL 幫助我們來查找錯誤。

以上一節的外連接為例說下怎麼列印最終 SQL

q = session \ .query(Book.id.label(&39;), Book.name.label(&39;), Category.id.label(&39;), Category.name.label(&39;)) \ .outerjoin(Category, Book.cat_id == Category.id) \ .filter(Book.id >= 9)raw_sql = q.statement \ .compile(compile_kwargs={&34;: True})print(raw_sql)

其中，q 為 sqlalchemy.orm.query.Query 類的對象。

結果

SELECT books.id AS book_id, books.name AS book_name, categories.id AS cat_id, categories.name AS cat_name FROM books LEFT OUTER JOIN categories ON books.cat_id = categories.id WHERE books.id >= 9

至此，SQLAlchemy ORM 常用的一些查詢方法和技巧已介紹完畢，希望能幫助到有需要的朋友。

原文連結：http://www.kevinbai.com/articles/30.html

關注公眾號「小小後端」獲取最新文章推送！