通過定義數(shù)據(jù)模型并使用 SQLAlchemy 的 ORM 中的會(huì)話對(duì)象，完全在 Python 中處理應(yīng)用程序的數(shù)據(jù)。
到目前為止，我們的 SQLAlchemy 之旅已經(jīng)涵蓋了管理數(shù)據(jù)庫連接和模型創(chuàng)建。然而，我們?nèi)绾螐臄?shù)據(jù)庫中提取我們想要的數(shù)據(jù)呢？
SQLAlchemy 的 ORM 查詢 API 簡化了我們編寫數(shù)據(jù)庫查詢的方式。我們可以通過將檢索數(shù)據(jù)的方法鏈接在一起來在 SQLAlchemy 會(huì)話上構(gòu)造查詢，而不是編寫原始 SQL 查詢。我們將深入研究 SQLAlchemy 的廣泛查詢 API，以了解查詢數(shù)據(jù)的所有方式。
創(chuàng)建 a Session
我們?cè)?a href="/diary/sql/579.html" target="_blank">上一篇文章中介紹了 SQLAlchemy 會(huì)話創(chuàng)建，并在之前的文章中解釋了引擎的概念。如果您跳過了這些帖子，請(qǐng)不要跳過。以下是復(fù)制+面食禮貌：
"""數(shù)據(jù)庫引擎和會(huì)話創(chuàng)建。""" from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker engine = create_engine( 'mysql+pymysql://user:password@host:3600/database', echo=True ) Session = sessionmaker(bind=engine) session = Session()
基本查詢語法
讓我們快速熟悉 SQLAlchemy 查詢 API 的基本結(jié)構(gòu)。SQLAlchemy會(huì)話對(duì)象有一個(gè)query()方法，它接受我們之前定義的數(shù)據(jù)模型的原始類。Customer下面是在模型上運(yùn)行的查詢的簡單開始；或者換句話說，對(duì)客戶SQL 表進(jìn)行查詢：
從 SQLAlchemy 會(huì)話中選擇記錄：
"""從 SQLAlchemy 會(huì)話構(gòu)建數(shù)據(jù)庫查詢。""" from .database import session from .models import Customer # ORM 查詢的示例結(jié)構(gòu) records = session .query(Customer) .FUNCTION()
在我們向鏈中添加另一種方法之前，調(diào)用.query(Customer)我們的會(huì)話并不是一個(gè)有效的查詢。所有會(huì)話查詢都以最終方法結(jié)束，以塑造/預(yù)測(cè)查詢結(jié)果：
all()將返回與我們的查詢匹配的所有記錄作為對(duì)象列表。如果我們?cè)谏厦娴牟樵冎惺褂胊llList[Customer] ，我們將收到 Python 數(shù)據(jù)類型的所有客戶記錄。
first()返回與查詢匹配的第一條記錄，無論有多少記錄與查詢匹配（“第一條”的構(gòu)成取決于表的排序方式）。這相當(dāng)于添加LIMIT 1到 SQL 查詢中。因此，要返回的 Python 類型將為Customer.
one()對(duì)于我們正在執(zhí)行的查詢最多應(yīng)存在一條記錄的情況（考慮按主鍵查詢）非常有用。在創(chuàng)建記錄之前驗(yàn)證記錄是否存在時(shí)，此語法特別有用。
scalar()如果存在則返回單個(gè)值，如果不存在值則返回None，或者如果返回多條記錄則引發(fā)異常。
get([VALUE(S)])搜索模型的主鍵以返回主鍵等于所提供的值的行。get()如果應(yīng)搜索多個(gè)外鍵，也接受元組。最后，還get()可以接受字典并返回列（字典鍵）與提供的值匹配的行。
要?jiǎng)?chuàng)建更復(fù)雜的查詢，我們將通過原始查詢上的鏈接方法添加到查詢中：
復(fù)雜的 SELECT 查詢：
"""從 SQLAlchemy 會(huì)話構(gòu)建數(shù)據(jù)庫查詢。""" from .database import session from .models import Customer # Example structure of an ORM query records = session .query(Customer) .METHOD_1() .METHOD_2() .FUNCTION()
查詢結(jié)果
如果我們執(zhí)行返回多條記錄的查詢，我們需要循環(huán)它們才能查看結(jié)果：
獲取所有客戶記錄并打印輸出：
"""從 SQLAlchemy 會(huì)話構(gòu)建數(shù)據(jù)庫查詢。""" from .database import session from .models import Customer # 獲取所有客戶記錄 records = session .query(Customer) .all() # 循環(huán)記錄 for record in records: print(record)
默認(rèn)情況下，SQLAlchemy ORM 將返回類的實(shí)例，這意味著上面的內(nèi)容將產(chǎn)生以下輸出：
會(huì)話查詢的輸出：
<Customer model 1> <Customer model 2> <Customer model 3> <Customer model 4> <Customer model 5> <Customer model 6> <Customer model 7> <Customer model 8> <Customer model 9> <Customer model 10>
如果您想獲取字典，請(qǐng)使用內(nèi)置__dict__方法：
以字典形式查看查詢結(jié)果：
... records = session .query(Customer) .all() for record in records: pp.pprint(record.__dict__)
相反，這會(huì)返回每行的字典對(duì)象：
會(huì)話查詢的輸出作為字典：
{ '_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x7f98c8221748>,
'email': 'kpaladini5i@senate.gov',
'first_name': 'Kenna',
'id': 199,
'join_date': datetime.datetime(2019, 4, 19, 0, 0),
'last_name': 'Paladini',
'preferred_language': 'Bulgarian'}
{ '_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x7f9918192d68>,
'email': 'rlebrun5j@narod.ru',
'first_name': 'Rriocard',
'id': 200,
'join_date': datetime.datetime(2015, 6, 8, 0, 0),
'last_name': 'Le Brun',
'preferred_language': 'Khmer'},
...當(dāng)然，您也可以創(chuàng)建自己的對(duì)象來只接收您想要/需要的列：
反序列化查詢結(jié)果：
...
# 獲取所有客戶
records = session.query(Customer).all()
# 循環(huán)記錄
for record in records:
recordObject = {
'name': record.name,
'position': record.position,
'team_name': record.team.name,
'team_city': record.team.city
}
print(recordObject)這輸出了一些更干凈的東西：
查詢反序列化的輸出：
{ 'email': 'kpaladini5i@senate.gov',
'first_name': 'Kenna',
'join_date': datetime.datetime(2019, 4, 19, 0, 0),
'last_name': 'Paladini',
'preferred_language': 'Bulgarian'}
{ 'email': 'rlebrun5j@narod.ru',
'first_name': 'Rriocard',
'join_date': datetime.datetime(2015, 6, 8, 0, 0),
'last_name': 'Le Brun',
'preferred_language': 'Khmer'},
...過濾結(jié)果
查詢中最常用的方法可能就是filter()方法。filter()相當(dāng)于 SQL WHERE子句，僅返回符合我們想要的條件的行：
選擇所有名為 Carl 的客戶：
... # 獲取“first_name”為“Carl”的記錄 records = session .query(Customer) .filter(Customer) .first_name == 'Carl') .all()
filter_by()
我們可以使用該方法編寫上述查詢，filter_by()如下所示：
過濾器filter_by：
... # 獲取“first_name”為“Carl”的記錄 records = session .query(Customer) .filter_by(first_name="Carl") .all()
與 不同的是filter()，filter_by()接受關(guān)鍵字參數(shù)（請(qǐng)注意此處語法的差異：filter()針對(duì)列對(duì)象檢查條件，而filter_by()查找與我們傳遞的參數(shù)匹配的列）。filter_by()只能搜索精確值，并作為簡單過濾查詢的一種簡寫。
like()
我們可以做的不僅僅是過濾簡單的條件。SQLAlchemy 有一個(gè)like()與 SQL 的工作方式等效的方法LIKE：
選擇名字以“J”開頭的客戶記錄：
...
# 獲取“first_name”以字母“J”開頭的記錄
records = session
.query(Customer)
.filter(Customer.first_name.like('J%'))
.all()正如預(yù)期的那樣，這將為我們提供客戶名字以J開頭的所有行：
{ 'email': 'jpugsley9@netvibes.com',
'first_name': 'Jarid',
'join_date': datetime.datetime(2017, 10, 11, 0, 0),
'last_name': 'Pugsley',
'preferred_language': 'Burmese'}
{ 'email': 'jdymockek@is.gd',
'first_name': 'Jeanna',
'join_date': datetime.datetime(2017, 11, 13, 0, 0),
'last_name': 'Dymocke',
'preferred_language': 'Malayalam'}
...高級(jí)查詢方法
除此之外filter()，還有一些我們應(yīng)該熟悉的基本方法。其中每一個(gè)都對(duì)應(yīng)于您可能熟悉的 SQL 關(guān)鍵字：
limit([INTEGER])：將行數(shù)限制為所提供的最大行數(shù)。
order_by([COLUMN])：按提供的列對(duì)結(jié)果進(jìn)行排序。
offset([INTEGER])：從第n行開始查詢。
下一部分涉及在模型之間執(zhí)行 JOIN 查詢，這要求我們首先定義模型上的關(guān)系。目前事情有點(diǎn)亂，因?yàn)槲覍?shí)際上要等到下一篇文章才會(huì)討論這個(gè)問題。抱歉打擾了，我正在努力！
連接不同表中的記錄并反序列化記錄
執(zhí)行 JOIN 和 UNION
我們之前已經(jīng)接觸過 JOIN，但我們即將將其提升一個(gè)檔次。我們正在使用兩種數(shù)據(jù)模型：一種用于客戶，一種用于訂單。每位顧客
連接不同表中的記錄并反序列化記錄：
...
import pprint
from .models import Order, Customer
pp = pprint.PrettyPrinter(indent=4)
# 對(duì) JOINed 表執(zhí)行 SELECT 查詢
records = session
.query(Customer)
.join(Order, Order.customer_id == Customer.id)
.all()
# 循環(huán)遍歷結(jié)果
for record in records:
record_object = {
'first_name': record.first_name,
'last_name': record.last_name,
'email': record.email,
'preferred_language': record.preferred_language,
'join_date': record.join_date,
'orders': []
}
for order in record.order:
order = {
'order_price': order.price,
'currency': order.currency,
'purchase_date': order.purchase_date,
'product': order.product
}
record_object['orders'].append(order)
pp.pprint(record_object)我們使用該方法執(zhí)行 JOIN join()。我們傳遞的第一個(gè)參數(shù)是我們將在“右側(cè)”加入的數(shù)據(jù)模型。然后，我們指定要“加入”的內(nèi)容：訂單模型的customer_id列和客戶模型的id列。
我們的外循環(huán)為我們提供了每個(gè)客戶，而我們的內(nèi)循環(huán)將每個(gè)訂單添加到適當(dāng)?shù)目蛻?。查看示例記錄?/p>
{ 'email': 'jtinline16@arizona.edu',
'first_name': 'Jerry',
'join_date': datetime.datetime(2016, 10, 27, 0, 0),
'last_name': 'Tinline',
'preferred_language': 'Icelandic',
'orders': [{'currency': 'IDR',
'order_price': 34.24,
'product': 'Beer - Corona',
'purchase_date': datetime.datetime(2019, 5, 5, 0, 0)},
{'currency': 'GEL',
'order_price': 25.75,
'product': 'Creamers - 10%',
'purchase_date': datetime.datetime(2019, 1, 27, 0, 0)}]}我們的朋友杰瑞有兩份訂單：一份是科羅娜啤酒，另一份是奶精。開始吧，杰瑞。
外連接
除了簡單的 JOIN 之外，我們還可以使用相同的語法執(zhí)行外部 JOIN：
... from .models import ExampleModel1, ExampleModel2 # 執(zhí)行外部 JOIN records = session .query(ExampleModel1) .outerjoin(ExampleModel2) .all()
Unions
我們還可以執(zhí)行 UNION 和 UNION ALL：
... from .models import ExampleModel1, ExampleModel2 # 執(zhí)行 UNION records = ExampleModel1.union(ExampleModel2)
要執(zhí)行全部聯(lián)合，只需替換union()為union_all()!
聚合函數(shù)和統(tǒng)計(jì)數(shù)據(jù)
與所有類似 SQL 的查詢語言一樣，我們也可以執(zhí)行一些聚合統(tǒng)計(jì)。我們可以使用以下內(nèi)容：
count([COLUMN])：計(jì)算列中的記錄數(shù)。
count(distinct([COLUMN]))：計(jì)算列中不同記錄的數(shù)量。
sum([COLUMN])：將列中的數(shù)值相加。
以下是我們?nèi)绾螆?zhí)行對(duì)列中的值進(jìn)行計(jì)數(shù)的查詢：
... from sqlalchemy import func # 計(jì)算具有“first_name”值的記錄數(shù) records = session .query(func.count(Customer.first_name)) .all() for record in records: print(record)
哪個(gè)輸出：
(200,)
可以輕松修改此查詢以僅計(jì)算不同值：
... from sqlalchemy import func from sqlalchemy import distinct # 計(jì)算不同“first_name”值的數(shù)量 records = session .query(func.count(distinct(Customer.first_name))) .all() for record in records: print(record)
使用 Group_by()
當(dāng)然，我們group_by()也可以在基于聚合的查詢上使用該方法。group_by()其工作原理與我們對(duì) SQL 和 Pandas 的期望類似：
“分組依據(jù)”聚合：
... # 執(zhí)行“GROUP BY”聚合查詢 records = session .query(func.count(Customer.first_name)) .group_by(Customer.first_name) .all()
突變（Mutations）
我們花了很多時(shí)間研究如何從數(shù)據(jù)庫中提取數(shù)據(jù)，但還沒有討論修改數(shù)據(jù)！今天我們議程上的最后一項(xiàng)是研究如何使用 SQLAlchemy ORM 添加、刪除和更改記錄。
插入行
我們添加數(shù)據(jù)的第一種方法是使用該add()方法。add()期望傳遞一個(gè)類的實(shí)例（特別是數(shù)據(jù)模型），并將創(chuàng)建一個(gè)新的數(shù)據(jù)庫行作為結(jié)果：
通過ORM插入記錄：
from .database import session from .models import Customer # Inserting records via data models customer = Customer( first_name='Todd', last_name='Birchard', email='fake@example.com', preferred_language='English', join_date=datetime.now() ) session.add(customer) session.commit()
添加數(shù)據(jù)的另一種方法是使用該insert()方法。與 不同的是add()，在 SQLAlchemy Tableinsert()對(duì)象上調(diào)用，并且不依賴于接收數(shù)據(jù)模型。不是 ORM 的一部分：insert()
... # 通過 SQLAlchemy `Table` 對(duì)象插入記錄 insert = [TABLE] .insert() .values( first_name='Todd', last_name='Jack Jones', email='fake@example.com', preferred_language='English', join_date=datetime.now() )
更新中 Updating
基于 的語法insert()，我們可以添加該update()方法來更改現(xiàn)有記錄的值。我們鏈接該where()方法來指定應(yīng)更新哪些行：
... # 通過 SQLAlchemy `Table` 對(duì)象更新記錄 result = [TABLE] .update() .where([TABLE].c.name == 'Todd') .values(email='newemail@example.com')
正在刪除 Deleting
在我們執(zhí)行的任何查詢中，我們可以附加delete()方法來刪除該查詢中包含的所有行（小心?。Ｏ旅鎰h除first_name列包含值“Carl”的所有記錄：
... # 刪除“first_name”為“Carl”的記錄 result = session .query(Customer) .filter(Customer.first_name == 'Carl') .delete()
delete()接受synchronize_session參數(shù)，該參數(shù)確定應(yīng)如何處理刪除：
False在提交會(huì)話之前不會(huì)執(zhí)行刪除。
'fetch'選擇要?jiǎng)h除的所有行并刪除匹配的行。
'evaluate'將評(píng)估當(dāng)前會(huì)話中的對(duì)象以確定應(yīng)刪除哪些行。文章來源：http://www.zghlxwxcb.cn/article/580.html
關(guān)鍵詞：SQLAlchemy，表達(dá)式語言，Python代碼查詢，數(shù)據(jù)庫查詢，會(huì)話對(duì)象文章來源地址http://www.zghlxwxcb.cn/article/580.html
到此這篇關(guān)于使用SQLAlchemy的表達(dá)式語言進(jìn)行高效Python代碼查詢的文章就介紹到這了,更多相關(guān)內(nèi)容可以在右上角搜索或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！
