之前做查詢一直覺得直接拼 SQL 比較方便瘦材，用了 SQLAlchemy 的 ORM 查詢之后，發(fā)現(xiàn)也還可以仿畸，還提高了可讀性食棕。
這篇文章主要說說 SQLAlchemy 常用的 ORM 查詢方式，偏實(shí)踐错沽〔鞠看了之后，對(duì)付開發(fā)中的查詢需求千埃，我覺得可以滿足不少憔儿。
為方便說明，假設(shè)有如下數(shù)據(jù)
圖書表 books
+----+--------+--------------------------+-------+
| id | cat_id | name | price |
+----+--------+--------------------------+-------+
| 1 | 1 | 生死疲勞 | 40.40 |
| 2 | 1 | 皮囊 | 31.80 |
| 3 | 2 | 半小時(shí)漫畫中國史 | 33.60 |
| 4 | 2 | 耶路撒冷三千年 | 55.60 |
| 5 | 2 | 國家寶藏 | 52.80 |
| 6 | 3 | 時(shí)間簡史 | 31.10 |
| 7 | 3 | 宇宙簡史 | 22.10 |
| 8 | 3 | 自然史 | 26.10 |
| 9 | 3 | 人類簡史 | 40.80 |
| 10 | 3 | 萬物簡史 | 33.20 |
+----+--------+--------------------------+-------+
分類表 categories
+----+--------------+
| id | name |
+----+--------------+
| 1 | 文學(xué) |
| 2 | 人文社科 |
| 3 | 科技 |
+----+--------------+
ORM 對(duì)象定義如下
注意：本文 Python 代碼在以下環(huán)境測(cè)試通過
- Python 3.6.0
- PyMySQL 0.8.1
- SQLAlchemy 1.2.8
# coding=utf-8
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Numeric
from sqlalchemy.orm import sessionmaker
Base = declarative_base()
engine = create_engine('mysql+pymysql://username:password'
'@127.0.0.1:3306/db_name?charset=utf8')
Session = sessionmaker(bind=engine)
session = Session()
def to_dict(self):
return {c.name: getattr(self, c.name, None)
for c in self.__table__.columns}
Base.to_dict = to_dict
class Book(Base):
__tablename__ = 'books'
id = Column(Integer, primary_key=True)
cat_id = Column(Integer)
name = Column('name', String(120))
price = Column('price', Numeric)
class Category(Base):
__tablename__ = 'categories'
id = Column(Integer, primary_key=True)
name = Column('name', String(30))
好了放可，下面進(jìn)入正題谒臼。
1 根據(jù)主鍵獲取記錄
當(dāng)我們獲取圖書的詳情時(shí)，很容易用到耀里。
book_id = 1
book = session.query(Book).get(book_id)
print(book and book.to_dict())
直接 get(primary_key) 就得到結(jié)果
{'id': 1, 'cat_id': 1, 'name': '生死疲勞',
'price': Decimal('40.40')}
當(dāng)然蜈缤，這樣也可以
book_id = 1
book = session.query(Book) \
.filter(Book.id == book_id) \
.first()
print(book and book.to_dict())
不過，還是前一種方式簡潔一些冯挎。
2 AND 查詢
我們最常用到的就是這種查詢底哥，比如我要獲取 cat_id 為 1 且價(jià)格大于 35 的書
books = session.query(Book) \
.filter(Book.cat_id == 1,
Book.price > 35) \
.all()
print([v.to_dict() for v in books])
執(zhí)行后，得到結(jié)果
[{'id': 1, 'cat_id': 1, 'name': '生死疲勞',
'price': Decimal('40.40')}]
filter() 里面的條件默認(rèn)是使用 AND 進(jìn)行連接房官，畢竟這最常用嘛趾徽。所以說，換成這樣用也是沒有問題的
from sqlalchemy import and_
books = session.query(Book) \
.filter(and_(Book.cat_id == 1,
Book.price > 35)) \
.all()
print([v.to_dict() for v in books])
不過翰守，通常來說孵奶，如果條件全是用 AND 連接的話，沒必要顯式的去用 and_蜡峰。
如果條件都是等值比較的話了袁，可以使用 filter_by() 方法，傳入的是關(guān)鍵字參數(shù)事示。
查詢 cat_id 等于 1 且價(jià)格等于 31.8 的圖書早像，可以這樣
books = session.query(Book) \
.filter_by(cat_id=1, price=31.8) \
.all()
print([v.to_dict() for v in books])
結(jié)果
[{'id': 2, 'cat_id': 1, 'name': '皮囊',
'price': Decimal('31.80')}]
這種方式相對(duì)于 filter() 來說僻肖，書寫要簡潔一些肖爵，不過條件都限制在了等值比較。
不同情況選擇合適的就好臀脏。
3 常用方法
除了上面使用的 get()劝堪、first()冀自、all() 外，還有下面的一些方法比較常用秒啦。
- one() 只獲取一條記錄熬粗，如果找不到記錄或者找到多條都會(huì)報(bào)錯(cuò)
# 找不到記錄會(huì)拋如下錯(cuò)誤
# sqlalchemy.orm.exc.NoResultFound: No row was found for one()
book = session \
.query(Book).filter(Book.id > 10) \
.one()
print(book and book.to_dict())
# 找到多條記錄會(huì)拋如下錯(cuò)誤
# sqlalchemy.orm.exc.MultipleResultsFound: Multiple rows were found for one()
book = session \
.query(Book).filter(Book.id < 10) \
.one()
print(book and book.to_dict())
# 正常，得到如下結(jié)果
# {'id': 10, 'cat_id': 3, 'name': '萬物簡史',
# 'price': Decimal('33.20')}
book = session \
.query(Book).filter(Book.id == 10) \
.one()
print(book and book.to_dict())
- count() 返回記錄條數(shù)
count = session \
.query(Book) \
.filter(Book.cat_id == 3) \
.count()
print(count)
結(jié)果
5
- limit() 限制返回的記錄條數(shù)
books = session \
.query(Book) \
.filter(Book.cat_id == 3) \
.limit(3) \
.all()
print([v.to_dict() for v in books])
結(jié)果
[{'id': 6, 'cat_id': 3, 'name': '時(shí)間簡史',
'price': Decimal('31.10')},
{'id': 7, 'cat_id': 3, 'name': '宇宙簡史',
'price': Decimal('22.10')},
{'id': 8, 'cat_id': 3, 'name': '自然史',
'price': Decimal('26.10')}]
- distinct() 與 SQL 的 distinct 語句行為一致
books = session \
.query(Book.cat_id) \
.distinct(Book.cat_id) \
.all()
print([dict(zip(v.keys(), v)) for v in books])
結(jié)果
[{'cat_id': 1}, {'cat_id': 2},
{'cat_id': 3}]
-
order_by()將記錄按照某個(gè)字段進(jìn)行排序
# 圖書按 ID 降序排列
# 如果要升序排列余境，去掉 .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])
結(jié)果
[{'id': 10, 'name': '萬物簡史'},
{'id': 9, 'name': '人類簡史'}]
- scalar() 返回調(diào)用 one() 后得到的結(jié)果的第一列值
book_name = session \
.query(Book.name) \
.filter(Book.id == 10)\
.scalar()
print(book_name)
結(jié)果
萬物簡史
- exist() 查看記錄是否存在
# 查看 ID 大于 10 的圖書是否存在
from sqlalchemy.sql import exists
is_exist = session \
.query(exists().where(Book.id > 10)) \
.scalar()
print(is_exist)
結(jié)果
False
4 OR 查詢
通過 OR 連接條件的情況也多驻呐，比如我要獲取 cat_id 等于 1 或者價(jià)格大于 35 的書
from sqlalchemy import or_
books = session.query(Book) \
.filter(or_(Book.cat_id == 1,
Book.price > 35)) \
.all()
print([v.to_dict() for v in books])
執(zhí)行，得到結(jié)果
[{'id': 1, 'cat_id': 1, 'name': '生死疲勞',
'price': Decimal('40.40')},
{'id': 2, 'cat_id': 1, 'name': '皮囊',
'price': Decimal('31.80')},
{'id': 4, 'cat_id': 2, 'name': '耶路撒冷三千年',
'price': Decimal('55.60')},
{'id': 5, 'cat_id': 2, 'name': '國家寶藏',
'price': Decimal('52.80')},
{'id': 9, 'cat_id': 3, 'name': '人類簡史',
'price': Decimal('40.80')}]
使用方式和 AND 查詢類似芳来，從 sqlalchemy 引入 or_含末，然后將條件放入就 OK 了。
5 AND 和 OR 并存的查詢
現(xiàn)實(shí)情況下即舌，我們很容易碰到 AND 和 OR 并存的查詢佣盒。比如，我現(xiàn)在要查詢價(jià)格大于 55 或者小于 25顽聂，同時(shí) cat_id 不等于 1 的圖書
from sqlalchemy import or_
books = session.query(Book) \
.filter(or_(Book.price > 55,
Book.price < 25),
Book.cat_id != 1) \
.all()
print([v.to_dict() for v in books])
結(jié)果
[{'id': 4, 'cat_id': 2, 'name': '耶路撒冷三千年',
'price': Decimal('55.60')},
{'id': 7, 'cat_id': 3, 'name': '宇宙簡史',
'price': Decimal('22.10')}]
又如肥惭，查詢圖書的數(shù)量，圖書滿足兩個(gè)要求中的一個(gè)即可：一是 cat_id 大于 5紊搪；二是 cat_id 小于 2 且價(jià)格大于 40蜜葱。可以這樣
from sqlalchemy import or_, and_
count = session.query(Book) \
.filter(or_(Book.cat_id > 5,
and_(Book.cat_id < 2,
Book.price > 40))) \
.count()
print(count)
結(jié)果
1
6 巧用列表或者字典的解包給查詢方法傳參
開發(fā)中耀石，我們經(jīng)常會(huì)碰到根據(jù)傳入的參數(shù)構(gòu)造查詢條件進(jìn)行查詢笼沥。比如
- 如果接收到非 0 的
cat_id，需要限制cat_id等于 0 - 如果接收到非 0 的 price娶牌，需要限制 price 等于傳入的 price
- 如果接收到非 0 的
min_price奔浅，需要限制 price 大于等于min_price - 如果接收到非 0 的
max_price，需要限制 price 小于等于max_price
我們就可以編寫類似的代碼
# 請(qǐng)求參數(shù)诗良，這里只是占位汹桦，實(shí)際由用戶提交的請(qǐng)求決定
params = {'cat_id': 1}
conditions = []
if params.get('cat_id', 0):
conditions.append(Book.cat_id == params['cat_id'])
if params.get('price', 0):
conditions.append(Book.price == params['price'])
if params.get('min_price', 0):
conditions.append(Book.price >= params['min_price'])
if params.get('max_price', 0):
conditions.append(Book.price <= params['max_price'])
books = session.query(Book).filter(*conditions).all()
print([v.to_dict() for v in books])
結(jié)果
[{'id': 1, 'cat_id': 1, 'name': '生死疲勞',
'price': Decimal('40.40')},
{'id': 2, 'cat_id': 1, 'name': '皮囊',
'price': Decimal('31.80')}]
OR 查詢類似，將列表解包傳給 or_() 即可鉴裹。
如果需求更復(fù)雜舞骆，AND 和 OR 都可能出現(xiàn)，這個(gè)時(shí)候根據(jù)情況多建幾個(gè)列表實(shí)現(xiàn)径荔。這里只向大家說明大致的思路督禽，就不舉具體的例子了。
當(dāng)然总处，如果都是等值查詢的話狈惫，比如只有這兩種情況
- 如果接收到非 0 的
cat_id，需要限制cat_id等于 0 - 如果接收到非 0 的 price鹦马，需要限制 price 等于傳入的 price
可以使用字典的解包給 filter_by() 傳參
# 請(qǐng)求參數(shù)胧谈，這里只是占位忆肾，實(shí)際由用戶提交的請(qǐng)求決定
params = {'price': 31.1}
condition_dict = {}
if params.get('cat_id', 0):
condition_dict['cat_id'] = params['cat_id']
if params.get('price', 0):
condition_dict['price'] = params['price']
books = session.query(Book) \
.filter_by(**condition_dict) \
.all()
print([v.to_dict() for v in books])
結(jié)果
[{'id': 6, 'cat_id': 3, 'name': '時(shí)間簡史',
'price': Decimal('31.10')}]
7 其它常用運(yùn)算符
除了上面看到的 ==、>菱肖、>=客冈、<、<=稳强、!= 之外场仲，還有幾個(gè)比較常用
- IN
# 查詢 ID 在 1、3退疫、5 中的記錄
books = session.query(Book) \
.filter(Book.id.in_([1, 3, 5])) \
.all()
- INSTR()
# 查詢名稱包含「時(shí)間簡史」的圖書
books = session.query(Book) \
.filter(Book.name.contains('時(shí)間簡史')) \
.all()
FIN_IN_SET()
# 查詢名稱包含「時(shí)間簡史」的圖書
# 這里顯然應(yīng)該用 INSTR() 的用法
# FIND_IN_SET() 一般用于逗號(hào)分隔的 ID 串查找
# 這里使用 FIND_IN_SET()燎窘，旨在說明用法
from sqlalchemy import func
books = session.query(Book) \
.filter(func.find_in_set('時(shí)間簡史', Book.name)) \
.all()
- LIKE
# 查詢名稱以「簡史」結(jié)尾的圖書
books = session.query(Book) \
.filter(Book.name.like('%簡史')) \
.all()
- NOT
上面的 IN、INSTR蹄咖、FIN_IN_SET褐健、LIKE 都可以使用 ~ 符號(hào)取反。比如
# 查詢 ID 不在 1 到 9 之間的記錄
books = session.query(Book) \
.filter(~Book.id.in_(range(1, 10))) \
.all()
8 查詢指定列
查詢名稱包含「簡史」的圖書的 ID 和名稱澜汤。如下
books = session.query(Book.id, Book.name) \
.filter(Book.name.contains('簡史')) \
.all()
print([dict(zip(v.keys(), v)) for v in books])
結(jié)果
[{'id': 6, 'name': '時(shí)間簡史'},
{'id': 7, 'name': '宇宙簡史'},
{'id': 9, 'name': '人類簡史'},
{'id': 10, 'name': '萬物簡史'}]
9 內(nèi)連接蚜迅、外連接
9.1 內(nèi)連接
獲取分類為「科技」，且價(jià)格大于 40 的圖書
# 如果 ORM 對(duì)象中定義有外鍵關(guān)系
# 那么 join() 中可以不指定關(guān)聯(lián)關(guān)系
# 否則俊抵，必須要
books = session \
.query(Book.id,
Book.name.label('book_name'),
Category.name.label('cat_name')) \
.join(Category, Book.cat_id == Category.id) \
.filter(Category.name == '科技',
Book.price > 40) \
.all()
print([dict(zip(v.keys(), v)) for v in books])
結(jié)果
[{'id': 9, 'book_name': '人類簡史',
'cat_name': '科技'}]
統(tǒng)計(jì)各個(gè)分類的圖書的數(shù)量
from sqlalchemy import func
books = session \
.query(Category.name.label('cat_name'),
func.count(Book.id).label('book_num')) \
.join(Book, Category.id == Book.cat_id) \
.group_by(Category.id) \
.all()
print([dict(zip(v.keys(), v)) for v in books])
結(jié)果
[{'cat_name': '文學(xué)', 'book_num': 2},
{'cat_name': '人文社科', 'book_num': 3},
{'cat_name': '科技', 'book_num': 5}]
9.2 外連接
為方便說明谁不，我們僅在這一小節(jié)中向 books 表中加入如下數(shù)據(jù)
+----+--------+-----------------+-------+
| id | cat_id | name | price |
+----+--------+-----------------+-------+
| 11 | 5 | 人性的弱點(diǎn) | 54.40 |
+----+--------+-----------------+-------+
查看 ID 大于等于 9 的圖書的分類信息
# outerjoin 默認(rèn)是左連接
# 如果 ORM 對(duì)象中定義有外鍵關(guān)系
# 那么 outerjoin() 中可以不指定關(guān)聯(lián)關(guān)系
# 否則，必須要
books = session \
.query(Book.id.label('book_id'),
Book.name.label('book_name'),
Category.id.label('cat_id'),
Category.name.label('cat_name')) \
.outerjoin(Category, Book.cat_id == Category.id) \
.filter(Book.id >= 9) \
.all()
print([dict(zip(v.keys(), v)) for v in books])
結(jié)果
[{'book_id': 9, 'book_name': '人類簡史',
'cat_id': 3, 'cat_name': '科技'},
{'book_id': 10, 'book_name': '萬物簡史',
'cat_id': 3, 'cat_name': '科技'},
{'book_id': 11, 'book_name': '人性的弱點(diǎn)',
'cat_id': None, 'cat_name': None}]
注意最后一條記錄徽诲。
10 打印 SQL
當(dāng)碰到復(fù)雜的查詢刹帕，比如有 AND、有 OR谎替、還有連接查詢時(shí)偷溺，有時(shí)可能得不到預(yù)期的結(jié)果，這時(shí)我們可以打出最終的 SQL 幫助我們來查找錯(cuò)誤钱贯。
以上一節(jié)的外連接為例說下怎么打印最終 SQL
q = session \
.query(Book.id.label('book_id'),
Book.name.label('book_name'),
Category.id.label('cat_id'),
Category.name.label('cat_name')) \
.outerjoin(Category, Book.cat_id == Category.id) \
.filter(Book.id >= 9)
raw_sql = q.statement \
.compile(compile_kwargs={"literal_binds": True})
print(raw_sql)
其中挫掏，q 為 sqlalchemy.orm.query.Query 類的對(duì)象。
結(jié)果
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 常用的一些查詢方法和技巧已介紹完畢尉共，希望能幫助到有需要的朋友。
查看最新文章請(qǐng)關(guān)注公眾號(hào)「小小后端」弃锐。
