SQLAlchemy 1.4 版後引入新的 2.0 查詢風格，簡化了資料函式庫操作流程。使用 Python 互動式環境搭配 SQLAlchemy，可以更直觀地進行資料函式庫查詢。首先，匯入必要的引擎、模型和會話物件，建立會話後，即可使用 select() 函式搭配模型類別查詢所有資料。透過 execute() 方法執行查詢，並使用 all()、first()、one() 等方法處理結果。此外，scalars() 方法適用於單值結果，並可結合其他方法簡化查詢。在實際應用中，通常使用迴圈遍歷結果集。
資料函式庫操作與查詢
在前面的章節中，我們已經成功地將產品資料匯入資料函式庫中。現在，是時候來學習如何對資料函式庫進行查詢操作了。如果你曾經使用過 SQLAlchemy，那麼你可能會注意到，從版本 1.4 開始，SQLAlchemy 對 ORM 查詢的構建方式進行了重大的更改。雖然舊的查詢實作仍然可用，但在本文中，我們將只使用新的查詢風格。SQLAlchemy 檔案將這種新的查詢風格稱為「2.0 查詢風格」，但這種查詢風格也可以在 1.4 版本中使用，只要在建立引擎和會話物件時設定 future=True 選項即可。
開啟 Python 互動式環境
最簡單的方式來操作新的資料函式庫是開啟一個 Python 互動式環境，在那裡你可以互動式地發出查詢。首先，你需要匯入引擎物件、模型類別和會話類別：
>>> from db import Session
>>> from models import Product
接著，你可以建立一個會話：
>>> session = Session()
這是一種不使用上下文管理器的建立會話的方式。在應用程式中，上下文管理器是一種非常方便的方式，但在 Python 互動式環境中，直接建立會話物件會更好。
查詢定義
關係型資料函式庫使用 SELECT 關鍵字來實作查詢。SQLAlchemy 提供了一個 select() 函式，具有類別似的功能。最簡單的查詢是傳回表中所有元素的查詢。以下是如何定義一個檢索資料函式庫中所有產品的查詢：
>>> from sqlalchemy import select
>>> q = select(Product)
select() 函式接受需要檢索的專案作為引數。當傳遞一個模型類別作為引數時，SQLAlchemy ORM 將檢索模型的所有屬性，並透明地傳回 Python 物件。
檢視 SQL 程式碼
有時，瞭解將要傳送到資料函式庫的 SQL 程式碼是非常有用的。與 SQLAlchemy 查詢物件相關聯的 SQL 程式碼可以在列印查詢時檢視：
>>> print(q)
SELECT products.id, products.name, products.manufacturer, products.year,
products.country, products.cpu
FROM products
在這裡，你可以看到如何將 Product 類別轉換為一個檢索表中所有屬性的 SELECT 陳述式。
執行查詢
在建立查詢物件之後，需要將其交給會話，會話將透過引擎維護的連線將其傳送到資料函式庫驅動程式執行。最通用的方式是使用會話的 execute() 方法：
>>> r = session.execute(q)
>>> list(r)
[(Product(1, "Acorn Atom"),), (Product(2, "BBC Micro"),), ..., (Product(149, "GEM 1000"),)]
execute() 方法傳回一個結果物件。這是一個可迭代的物件，可以檢索查詢結果。在上面的例子中，可迭代物件被轉換為列表，以強制檢索和顯示所有結果。
結果處理方法
結果物件具有多種方法，可以用來檢索查詢結果：
all(): 傳回所有結果行。first(): 傳回第一個結果行，如果沒有結果，則傳回None。如果有多於一行的結果，其他行將被丟棄。one(): 傳回第一個且唯一的結果行，如果沒有結果或有多於一個結果，則引發異常。one_or_none(): 傳回第一個且唯一的結果行，如果沒有結果，則傳回None。如果有多於一個結果，則引發異常。
使用 scalars() 方法
如果你知道每個結果行只包含一個值，那麼你可以使用 scalars() 方法來執行查詢：
>>> session.scalars(q).all()
[Product(1, "Acorn Atom"), Product(2, "BBC Micro"), ..., Product(149, "GEM 1000")]
使用 scalars()，SQLAlchemy 傳回一個不同的「結果」物件，它只迭代每個結果行的第一個值。如果查詢傳回每個結果行的多個值，那麼其他值將被丟棄。
結果遍歷
在應用程式中，通常會在 for 迴圈中遍歷結果：
>>> r = session.execute(q)
>>> for row in r:
... print(row)
(Product(1, "Acorn Atom"),)
(Product(2, "BBC Micro"),)
...
(Product(149, "GEM 1000"),)
注意每個結果的結構。SQLAlchemy 將每個結果作為元組傳回，因為查詢有時會傳回每個行的多個值。
簡化查詢方法
為了方便起見，SQLAlchemy 提供了一些額外的查詢執行方法，它們結合了 scalars() 和其他方法。例如：
scalar(q)等同於scalars(q).first()，傳回第一個結果行的第一個值，如果沒有結果，則傳回None。scalar_one(q)等同於scalars(q).one()，傳回唯一結果行的第一個值，如果沒有結果或有多於一個結果，則引發異常。scalar_one_or_none(q)等同於scalars(q).one_or_none()，傳回唯一結果行的第一個值，如果沒有結果，則傳回None，如果有多於一個結果，則引發異常。
查詢過濾與排序
在前面的例子中，我們已經看到了如何使用 select() 函式來檢索資料函式庫中的資料。然而，在許多情況下，我們需要對查詢結果進行過濾，以取得特定的資料子集。本文將介紹如何使用 SQLAlchemy 來實作查詢過濾。
使用 where() 方法進行過濾
SQLAlchemy 提供了 where() 方法來為查詢新增過濾條件。下面的例子展示瞭如何檢索由 Commodore 製造的產品：
>>> q = select(Product).where(Product.manufacturer == 'Commodore')
>>> session.scalars(q).all()
[Product(39, "PET"), Product(40, "VIC-20"), ..., Product(48, "Amiga")]
內容解密：
where()方法用於新增查詢條件。Product.manufacturer == 'Commodore'定義了一個條件：只檢索製造商為 Commodore 的產品。
多重過濾條件
可以透過多次呼叫 where() 方法或在單次呼叫中傳入多個條件來應用多重過濾。下面的例子展示瞭如何檢索 Commodore 在 1980 年製造的產品：
>>> q = (select(Product)
... .where(Product.manufacturer == 'Commodore')
... .where(Product.year == 1980))
或
>>> q = select(Product).where(Product.manufacturer == 'Commodore', Product.year == 1980)
內容解密：
- 多次呼叫
where()方法與單次呼叫傳入多個條件的效果相同。 - 多個條件之間預設使用邏輯 AND 運算子。
邏輯運算子
SQLAlchemy 提供了 or_()、and_() 和 not_() 函式來支援邏輯運算。下面的例子展示瞭如何檢索在 1970 年之前或 1990 年之後製造的產品：
>>> from sqlalchemy import or_
>>> q = select(Product).where(or_(Product.year < 1970, Product.year > 1990))
內容解密：
or_()函式用於將多個條件以邏輯 OR 運算子結合。- 其他邏輯運算子如
and_()和not_()也可按需使用。
LIKE 運算子
LIKE 運算子可用於實作簡單的搜尋功能。下面的例子展示瞭如何檢索名稱中包含 “Sinclair” 的產品：
>>> q = select(Product).where(Product.name.like('%Sinclair%'))
內容解密：
like()方法接受一個模式字串，支援%和_萬用字元。%用於匹配零個或多個字元，而_用於匹配單個字元。
BETWEEN 運算子
可以使用 between() 方法來檢索某一範圍內的資料。下面的例子展示瞭如何檢索在 1970 年代製造的產品：
>>> q = select(Product).where(Product.year.between(1970, 1979))
內容解密：
between()方法定義了一個範圍，用於篩選出落在該範圍內的資料。
查詢生成的 SQL 陳述式
可以透過列印查詢物件來檢視 SQLAlchemy 生成的 SQL 陳述式。下面的例子展示瞭如何檢視查詢的 SQL 陳述式，以及如何啟用 literal_binds 以顯示實際的引數值：
>>> print(q)
SELECT products.id, products.name, products.manufacturer, products.year,
products.country, products.cpu
FROM products
WHERE products.year BETWEEN :year_1 AND :year_2
>>> print(q.compile(compile_kwargs={'literal_binds': True}))
SELECT products.id, products.name, products.manufacturer, products.year,
products.country, products.cpu
FROM products
WHERE products.year BETWEEN 1970 AND 1979
內容解密：
- SQLAlchemy 自動使用引數化查詢以防止 SQL 注入攻擊。
- 可以透過
compile_kwargs={'literal_binds': True}檢視實際的引數值，但這不應在生產環境中使用。
資料排序與檢索
在資料函式庫查詢中，結果的排序是非常重要的一環。SQLAlchemy 提供了 order_by() 方法來指定查詢結果的排序順序。以下範例展示如何使用此方法按產品名稱的字母順序檢索產品：
q = select(Product).order_by(Product.name) session.scalars(q).all() [Product(10, “464 Plus”), Product(11, “6128 Plus”), …, Product(127, “ZX Spectrum”)]
若要進行反向排序，可以在指定的欄位屬性上呼叫 desc() 方法。以下範例按產品的製造年份進行降序排序，將最新的產品排在前面：
q = select(Product).order_by(Product.year.desc()) session.scalars(q).all() [Product(6, “A7000”), Product(33, “Falcon”), …, Product(74, “Honeywell 316”)]
在許多情況下，單一的排序條件可能不足以滿足需求。例如，在上述範例中，同一年份內製造的電腦產品會以任意順序傳回。order_by() 方法可以接受多個引數，每個引數都會新增一層排序條件。以下範例對產品先按年份降序排序，然後按產品名稱升序排序：
q = select(Product).order_by(Product.year.desc(), Product.name.asc())
請注意，asc() 方法用於指定升序排序，雖然升序是預設的，但有時明確指出排序方向可以使查詢陳述式更清晰。
存取個別欄位
在前面的範例中，查詢結果都是整個資料列，對應到 ORM 的實體類別例項。雖然查詢 ORM 實體是一種常見的做法，但 select() 函式非常靈活，可以處理更細粒度的資料。
例如，應用程式可能只需要檢索某個特定的欄位。以下查詢只檢索產品的名稱：
q = select(Product.name) session.scalars(q).all() [‘Acorn Atom’, ‘BBC Micro’, …, ‘GEM 1000’]
select() 函式不限於檢索每個結果列中的單一值，也可以同時請求多個值。以下查詢取得每個產品的名稱和製造商：
q = select(Product.name, Product.manufacturer) session.execute(q).all() [(‘Acorn Atom’, ‘Acorn Computers Ltd’), (‘BBC Micro’, ‘Acorn Computers Ltd’), …]
注意，這個範例必須使用 session.execute() 來執行查詢，以便將每列結果中的值對傳回為元組。
聚合函式
select() 函式也可以與 SQL 函式一起使用，這些函式會在檢索資料時動態轉換資料。其中一個非常有用的函式是 count()，它會將所有結果列替換為結果的總數。以下範例計算資料函式庫中儲存的產品數量：
from sqlalchemy import func q = select(func.count(Product.id)) r = session.scalar(q) r 149
上述範例中使用的 count() 函式將結果列表簡化為單一值，因此使用 scalar() 方法來檢索結果。在這個例子中，使用 Product.id 作為 count() 的引數是任意的，給定 Product 類別的任何欄位屬性都會得到相同的結果，因為實際資料並不重要。
還有另一種形式的查詢，不需要選取隨機欄位來計算結果數量：
q = select(func.count()).select_from(Product) r = session.scalar(q) r 149
在第二種形式中，count() 函式沒有給定任何引數，以表示需要計算結果數量，而不指定要計數的資料。使用這種格式時，必須新增 select_from() 方法來組態查詢中使用的表格，因為 SQLAlchemy 無法從給定 select() 函式的引數中自動確定它。
另兩個有用的 SQL 函式是 min() 和 max()。以下範例傳回資料函式庫中產品製造的最早和最晚年份：
q = select(func.min(Product.year), func.max(Product.year)) r = session.execute(q) r.first() (1969, 1995)
這個查詢必須使用 execute() 因為它檢索每列兩個值。min() 和 max() 函式將結果列表簡化為單一行，因此沒有必要像前面的範例那樣使用 all() 來檢索結果。當預先知道只有一行結果時，使用 first() 或 one() 方法更方便，其中後者在查詢傳回非單一結果列時會引發例外。
結果分組
目前形式的資料函式庫只有產品作為第一類別實體，但有時應用程式可能對檢索相關資料屬性（如製造商）感興趣。以下是嘗試從這個表格中取得電腦製造商列表：
q = select(Product.manufacturer).order_by(Product.manufacturer) session.scalars(q).all() [‘Acorn Computers Ltd’, ‘Acorn Computers Ltd’, …, ‘West Computer AS’]
但顯然，這個查詢有一個問題。即使查詢只檢索製造商，查詢的表格中包含的是產品，因此每個結果列對應一個產品，而擁有多個產品的製造商會被重複列出。
內容解密：
此章節主要講解了SQLAlchemy中如何進行資料排序、存取個別欄位、使用聚合函式以及結果分組。首先，我們瞭解到可以使用order_by()方法對查詢結果進行排序，並且可以透過呼叫desc()或asc()方法來指定排序的方向。其次，我們學習瞭如何使用select()函式存取個別欄位，以及如何同時檢索多個欄位。此外，我們還瞭解瞭如何使用SQL函式，如count(), min(), 和 max()，來動態轉換資料。最後，我們討論了結果分組的問題，並指出在目前的資料函式庫設計中，直接從產品表檢索製造商列表會導致重複項。
# 以下是一個綜合範例，展示瞭如何使用SQLAlchemy進行資料排序、存取個別欄位和使用聚合函式
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# 定義資料函式庫連線和模型
engine = create_engine('sqlite:///example.db')
Base = declarative_base()
class Product(Base):
__tablename__ = 'products'
id = Column(Integer, primary_key=True)
name = Column(String)
manufacturer = Column(String)
year = Column(Integer)
# 建立表格和Session
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
# 新增一些示例資料
# ...
# 按名稱排序檢索產品
q = session.query(Product).order_by(Product.name)
products = q.all()
for product in products:
print(product.name)
# 只檢索產品名稱
q = session.query(Product.name)
product_names = q.all()
for name in product_names:
print(name)
# 使用count()函式計算產品數量
from sqlalchemy import func
q = session.query(func.count(Product.id))
count = q.scalar()
print(f"Total products: {count}")
# 使用min()和max()函式找出最早和最晚的製造年份
q = session.query(func.min(Product.year), func.max(Product.year))
result = q.first()
print(f"Earliest year: {result[0]}, Latest year: {result[1]}")
內容解密：
- 匯入必要的模組：首先，我們需要匯入SQLAlchemy中的必要模組，包括
create_engine、Column、Integer、String、declarative_base和sessionmaker。 - 定義資料函式庫模型：我們定義了一個名為
Product的類別，它繼承自declarative_base()傳回的基礎類別，並定義了四個欄位：id、name、manufacturer和year。 - 建立資料函式庫連線和Session：我們建立了一個SQLite資料函式庫引擎，並建立了
Product類別對應的表格。同時，我們建立了一個Session類別，用於與資料函式庫進行互動。 - 新增示例資料：雖然這裡沒有顯示具體的新增程式碼，但通常我們會透過建立
Product例項並將其新增到Session中來新增資料。 - 按名稱排序檢索產品：我們使用
session.query(Product).order_by(Product.name)來按名稱升序檢索所有產品，並遍歷結果列印出每個產品的名稱。 - 只檢索產品名稱：透過修改查詢只選擇
Product.name，我們能夠僅檢索產品名稱。 - 計算產品數量：利用
func.count(Product.id)，我們能夠計算出資料函式庫中的產品總數。 - 找出最早和最晚的製造年份：透過
func.min(Product.year)和func.max(Product.year)，我們能夠找出資料函式庫中記錄的最早和最晚的產品製造年份。