SQLAlchemy 提供了 ORM 的便利性，但某些資料函式庫操作，例如 PostgreSQL 的右外連線，需要一些技巧才能實作。理解 SQLAlchemy 如何處理外連線，並配合資料函式庫特性調整查詢策略，才能有效率地擷取所需資料。本文除了探討 PostgreSQL 外連線的 SQLAlchemy 實作技巧外，也涵蓋了多語言部落格文章的資料函式庫模型設計，包含語言模型的建立、自參考關聯的應用，以及如何使用 Alembic 進行資料函式庫遷移。此外，文章也提供了多語言文章匯入指令碼的實務解析，以及如何使用 SQLAlchemy 進行多語言文章的統計查詢，包含各語言文章數量、原文與譯文分開統計、以及原文翻譯數量統計等實務案例。
PostgreSQL 外連線（Outer Join）與 SQLAlchemy 的應用
PostgreSQL 支援所有型別的外連線（outer join），但 SQLAlchemy 只實作了完全外連線（full outer join）和左外連線（left outer join）。當需要使用右外連線（right outer join）時，必須交換表格順序以使用左外連線。
將內連線轉換為完全外連線
在前面的查詢中，將第一個連線更改為完全外連線將確保檢索到所有部落格文章，而不僅僅是那些可以與產品匹配的文章。然後，下一個與 BlogView 的連線不會丟棄任何頁面瀏覽量。將預設的內連線轉換為完全外連線的唯一更改是在 join() 子句中新增 full=True 引數：
q = (select(Product.name, page_views)
.join(Product.blog_articles, full=True)
.join(BlogArticle.views)
.where(BlogView.timestamp.between(
datetime(2022, 11, 1), datetime(2022, 12, 1)))
.group_by(Product)
.order_by(page_views.desc()))
內容解密：
.join(Product.blog_articles, full=True)：使用完全外連線檢索所有產品和部落格文章。.join(BlogArticle.views)：預設為內連線，將部落格文章與其瀏覽量進行匹配。.where(BlogView.timestamp.between(...))：篩選特定時間範圍內的頁面瀏覽量。
執行此查詢後，結果將包括產品名稱和頁面瀏覽量，其中包含一個 None 產品名稱的條目，對應於未與任何產品相關聯的部落格文章的頁面瀏覽量。
使用右外連線的替代方案
由於 SQLAlchemy 不直接支援右外連線，可以透過反轉連線方向並使用左外連線來實作相同的效果。這可以透過在 join() 子句中新增 isouter=True 引數來完成：
q = (select(Product.name, page_views)
.join(BlogArticle.product, isouter=True)
.join(BlogArticle.views)
.where(BlogView.timestamp.between(
datetime(2022, 11, 1), datetime(2022, 12, 1)))
.group_by(Product)
.order_by(page_views.desc()))
內容解密：
.join(BlogArticle.product, isouter=True)：使用左外連線檢索所有部落格文章及其相關產品，如果沒有相關產品，則產品名稱為None。.join(BlogArticle.views)：將部落格文章與其瀏覽量進行匹配。.where(BlogView.timestamp.between(...))：篩選特定時間範圍內的頁面瀏覽量。
為什麼完全外連線和右外連線查詢傳回相同的結果？
乍一看，完全外連線應該傳回更多資料，因為它應該包括所有產品和所有部落格文章，無論它們是否匹配。然而，在這個查詢中，第二個連線是內連線，它將產品和部落格文章的配對與 BlogView 記錄進行匹配，這樣，所有 (Product, None) 對都被丟棄，因為 None 無法與任何 BlogView 記錄匹配。
改進查詢以保留所有產品
為了保留所有產品，包括那些沒有部落格文章或沒有頁面瀏覽量的產品，第二個連線也需要升級為完全外連線。同時，where() 子句需要更新以允許 BlogView.timestamp 為 None 的情況：
from sqlalchemy import or_
q = (select(Product.name, page_views)
.join(Product.blog_articles, full=True)
.join(BlogArticle.views, full=True)
.where(or_(
BlogView.timestamp == None,
BlogView.timestamp.between(
datetime(2022, 11, 1), datetime(2022, 12, 1))))
.group_by(Product)
.order_by(page_views.desc(), Product.name))
內容解密：
.join(Product.blog_articles, full=True)和.join(BlogArticle.views, full=True)：使用完全外連線檢索所有相關資料。.where(or_(...))：篩選特定時間範圍內的頁面瀏覽量，或允許None值。
在不支援完全外連線的資料函式庫中的解決方案
在不支援完全外連線的資料函式庫中，可以透過執行兩個查詢來模擬完全外連線的效果：一個用於檢索具有頁面瀏覽量的產品，另一個用於檢索沒有頁面瀏覽量的產品。
q2 = (select(Product.name, page_views)
.join(Product.blog_articles, isouter=True)
.join(BlogArticle.views, isouter=True)
.where(or_(
BlogView.timestamp == None,
BlogView.timestamp.between(
datetime(2022, 11, 1), datetime(2022, 12, 1))))
.group_by(Product)
.having(page_views == 0)
.order_by(Product.name))
內容解密：
.join(Product.blog_articles, isouter=True)和.join(BlogArticle.views, isouter=True)：使用左外連線檢索所有相關資料。.where(or_(...))：篩選特定時間範圍內的頁面瀏覽量，或允許None值。.having(page_views == 0)：僅檢索頁面瀏覽量為零的產品。
多語言部落格文章的管理與分析
在 RetroFun 的網站分析專案中，第三階段的主要目標是擴充資料函式庫以追蹤每篇文章的語言，並識別哪些文章是其他文章的翻譯版本。這一階段的變更將使資料函式庫能夠支援多語言內容，並提供更多有價值的報表。
語言模型的建立與關聯
首先，需要新增一個 Language 模型，並在 BlogArticle 模型中建立一個一對多的關聯。這使得每種語言可以對應多篇文章。
程式碼實作：新增語言模型與關聯
class BlogArticle(Model):
# ...
language_id: Mapped[Optional[int]] = mapped_column(
ForeignKey('languages.id'), index=True)
# ...
language: Mapped[Optional['Language']] = relationship(
back_populates='blog_articles')
# ...
class Language(Model):
__tablename__ = 'languages'
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(32), index=True, unique=True)
blog_articles: WriteOnlyMapped['BlogArticle'] = relationship(
back_populates='language')
def __repr__(self):
return f'Language({self.id}, "{self.name}")'
內容解密：
Language模型：定義了一個新的資料表languages，包含id和name欄位，分別代表語言的唯一識別碼和語言名稱。BlogArticle中的language_id：新增了一個外部索引鍵欄位language_id，與languages表的id關聯，用於標識每篇文章的語言。language關聯屬性：透過 SQLAlchemy 的relationship，在BlogArticle中建立了與Language的關聯，使得每篇文章可以對應到一種語言。
自參考關聯：處理文章翻譯
為了追蹤文章之間的翻譯關係，需要在 BlogArticle 模型中建立一個自參考關聯。這是因為一篇文章可以有多個翻譯版本，而每個翻譯版本本身也是一篇文章。
程式碼實作：自參考關聯
class BlogArticle(Model):
# ...
translation_of_id: Mapped[Optional[int]] = mapped_column(
ForeignKey('blog_articles.id'), index=True)
# ...
translation_of: Mapped[Optional['BlogArticle']] = relationship(
remote_side=id, back_populates='translations')
translations: Mapped[list['BlogArticle']] = relationship(
back_populates='translation_of')
# ...
內容解密：
translation_of_id外部索引鍵：新增了一個外部索引鍵欄位translation_of_id，它參考了blog_articles表自身的id欄位，用於標識一篇文章的原始版本。translation_of和translations關聯屬性：translation_of：表示某篇文章是另一個文章的翻譯版本，透過remote_side=id引數，SQLAlchemy 能夠正確理解這層關係。translations：表示某篇文章擁有的所有翻譯版本。
- 自參考關聯的組態：由於這是一個自參考關聯，需要小心組態
remote_side引數，以消除 SQLAlchemy 在解析關係時的歧義。
資料函式庫遷移
完成模型變更後，需要透過 Alembic 生成並執行資料函式庫遷移指令碼，以更新資料函式庫結構。
(venv) $ alembic revision --autogenerate -m "multi-language support"
(venv) $ alembic upgrade head
內容解密：
alembic revision --autogenerate：自動生成一個新的資料函式庫遷移指令碼，根據模型變更產生相應的 SQL 陳述式。alembic upgrade head：將生成的遷移指令碼應用到資料函式庫，完成資料函式庫結構的更新。
多語言支援匯入與查詢實務解析
語言與翻譯關係匯入指令碼解析
為實作多語言支援，專案採用了import_languages.py指令碼來匯入語言及翻譯關係。該指令碼主要處理articles.csv檔案中的language和translation_of欄位。
程式碼實作重點
import csv
from sqlalchemy import select
from db import Session
from models import BlogArticle, Language
def main():
with Session() as session:
with session.begin():
# 快取機制提升查詢效率
all_articles = {}
all_languages = {}
with open('articles.csv') as f:
reader = csv.DictReader(f)
for row in reader:
# 取得或建立文章物件
article = all_articles.get(row['title'])
if article is None:
article = session.scalar(select(BlogArticle).where(
BlogArticle.title == row['title']))
all_articles[article.title] = article
# 處理語言資訊
language = all_languages.get(row['language'])
if language is None:
language = session.scalar(select(Language).where(
Language.name == row['language']))
if language is None:
language = Language(name=row['language'])
session.add(language)
all_languages[language.name] = language
article.language = language
# 處理翻譯關係
if row['translation_of']:
translation_of = all_articles.get(row['translation_of'])
if translation_of is None:
translation_of = session.scalar(select(
BlogArticle).where(BlogArticle.title ==
row['translation_of']))
all_articles[article.title] = article
article.translation_of = translation_of
if __name__ == '__main__':
main()
#### 內容解密：
- 快取機制的應用：使用
all_articles和all_languages字典來快取已載入的文章和語言物件，顯著減少資料函式庫查詢次數。 - 語言處理邏輯：首先檢查語言是否已存在，若不存在則建立新的
Language例項並加入資料函式庫。 - 翻譯關係建立：透過
translation_of欄位建立原文與譯文之間的關聯，若文章為譯文則指定其原文。 - 資料函式庫操作最佳化：使用
with session.begin()確保資料函式庫操作的原子性。
語言查詢實務
在完成語言和翻譯關係的匯入後，可以透過SQLAlchemy進行各種查詢操作。
各語言文章數量統計
q = (select(Language, func.count(BlogArticle.id))
.join(Language.blog_articles)
.group_by(Language)
.order_by(Language.name))
session.execute(q).all()
#### 內容解密：
- 查詢邏輯：透過
Language與BlogArticle的關聯進行分組統計。 - 排序方式：按語言名稱進行排序。
- 統計結果：取得各語言的文章總數。
原文與譯文分開統計
# 原文統計
q = (select(Language, func.count(BlogArticle.id))
.join(Language.blog_articles)
.where(BlogArticle.translation_of == None)
.group_by(Language)
.order_by(Language.name))
# 譯文統計
q = (select(Language, func.count(BlogArticle.id))
.join(Language.blog_articles)
.where(BlogArticle.translation_of != None)
.group_by(Language)
.order_by(Language.name))
#### 內容解密：
- 篩選條件：使用
translation_of欄位判斷文章是否為原文或譯文。 - 特別注意：SQLAlchemy中與
None的比較必須使用==或!=運算元。 - 查詢結果：分別取得各語言的原文和譯文數量。
原文翻譯數量統計
TranslatedBlogArticle = aliased(BlogArticle)
article_count = func.count(TranslatedBlogArticle.id).label(None)
q = (select(BlogArticle, article_count)
.join(TranslatedBlogArticle.translation_of)
.group_by(BlogArticle)
.order_by(article_count.desc(), BlogArticle.title))
#### 內容解密：
- 別名機制：使用
aliased函式為BlogArticle建立別名，以解決自關聯查詢的歧義性。 - 統計邏輯：透過分組計數取得每篇原文的翻譯數量。
- 排序方式：按翻譯數量降序排列，若數量相同則按標題排序。
