SQLAlchemy 資料函式庫查詢技術詳解

SQLAlchemy 作為 Python 的 ORM 工具，提供簡潔易用的介面來操作資料函式庫。本文除了涵蓋基本查詢外，更探討了關聯查詢、聚合函式、排序、分頁等進階技巧，並搭配左外連線、子查詢等方法，解決複雜的資料查詢需求。同時也介紹瞭如何使用 Alembic 進行資料函式庫遷移，確保資料函式庫結構與程式碼同步更新，提升開發效率。

產品評論相關查詢

1. 產品平均評分與評論數量

要取得產品的平均評分和評論數量，並按評論數量降序排序，我們可以使用以下查詢：

product_rating = func.avg(ProductReview.rating).label(None)
review_count = func.count(ProductReview.rating).label(None)
q = (select(Product, product_rating, review_count)
     .join(Product.reviews)
     .group_by(Product)
     .order_by(review_count.desc()))
session.execute(q).all()

內容解密：

• 使用 func.avg 和 func.count 來計算平均評分和評論數量。
• join(Product.reviews) 將產品與其評論進行連線。
• group_by(Product) 按照產品進行分組。
• order_by(review_count.desc()) 按評論數量降序排序。

若要包含沒有評論的產品，需改用左外連線：

q = (select(Product, product_rating, review_count)
     .join(Product.reviews, isouter=True)
     .group_by(Product)
     .order_by(review_count.desc()))
session.execute(q).all()

內容解密：

• isouter=True 將內連線改為左外連線，以包含沒有評論的產品。

部落格文章瀏覽量查詢

1. 2020 年 3 月瀏覽量超過 40 次的部落格文章

q = (select(BlogArticle)
     .join(BlogArticle.views)
     .where(BlogView.timestamp.between(datetime(2020, 3, 1), datetime(2020, 4, 1)))
     .group_by(BlogArticle)
     .having(func.count(BlogView.id) > 50))
session.execute(q).all()

內容解密：

• 使用 between 方法篩選特定時間範圍內的瀏覽記錄。
• having(func.count(BlogView.id) > 50) 篩選瀏覽次數超過 50 次的文章。

(以下省略 SQLAlchemy 技術索引的程式碼與說明)

SQLAlchemy 查詢範例詳解

本文將探討一系列使用 SQLAlchemy 進行資料函式庫查詢的範例，涵蓋產品評論、部落格文章瀏覽量等主題。透過這些範例，我們將展示如何利用 SQLAlchemy 的強大功能來實作複雜的資料查詢。

產品評論相關查詢

1. 產品平均評分與評論數量

要取得產品的平均評分和評論數量，並按評論數量降序排序，我們可以使用以下查詢：

product_rating = func.avg(ProductReview.rating).label(None)
review_count = func.count(ProductReview.rating).label(None)
q = (select(Product, product_rating, review_count)
     .join(Product.reviews)
     .group_by(Product)
     .order_by(review_count.desc()))
session.execute(q).all()

內容解密：

• 使用 func.avg 和 func.count 來計算平均評分和評論數量。
• join(Product.reviews) 將產品與其評論進行連線。
• group_by(Product) 按照產品進行分組。
• order_by(review_count.desc()) 按評論數量降序排序。

若要包含沒有評論的產品，需改用左外連線：

q = (select(Product, product_rating, review_count)
     .join(Product.reviews, isouter=True)
     .group_by(Product)
     .order_by(review_count.desc()))
session.execute(q).all()

內容解密：

• isouter=True 將內連線改為左外連線，以包含沒有評論的產品。

部落格文章瀏覽量查詢

1. 2020 年 3 月瀏覽量超過 40 次的部落格文章

q = (select(BlogArticle)
     .join(BlogArticle.views)
     .where(BlogView.timestamp.between(datetime(2020, 3, 1), datetime(2020, 4, 1)))
     .group_by(BlogArticle)
     .having(func.count(BlogView.id) > 50))
session.execute(q).all()

內容解密：

• 使用 between 方法篩選特定時間範圍內的瀏覽記錄。
• having(func.count(BlogView.id) > 50) 篩選瀏覽次數超過 50 次的文章。

資料分析與呈現

使用 Plantuml 圖表呈現查詢邏輯

@startuml
skinparam backgroundColor #FEFEFE
skinparam defaultTextAlignment center
skinparam rectangleBackgroundColor #F5F5F5
skinparam rectangleBorderColor #333333
skinparam arrowColor #333333

title 使用 Plantuml 圖表呈現查詢邏輯

rectangle "一對多" as node1
rectangle "多對一" as node2

node1 --> node2

@enduml

此圖示展示了產品、評論、製造商和國家之間的關聯。

SQLAlchemy 技術索引

SQLAlchemy 是 Python 中最為廣泛使用的 ORM（Object-Relational Mapping）工具之一，提供了一套完整的資料函式庫操作介面。本文將根據提供的索引內容，探討 SQLAlchemy 的核心功能、組態、操作方法及其在不同場景下的應用。

核心功能與元件

1. ORM 與 Core 模組

SQLAlchemy 將其功能分為 ORM 和 Core 兩個主要模組。ORM 提供了一種將資料函式庫表對映為 Python 類別的方法，而 Core 則提供了底層的 SQL 表示式語言和資料函式庫連線功能。

2. Engine 與 Session

• Engine: 是 SQLAlchemy 中的核心連線池和資料函式庫方言處理元件，負責管理資料函式庫連線。
• Session: 是 ORM 中的主要介面，用於管理物件的生命週期，執行查詢和變更資料函式庫操作。

資料函式庫操作與查詢

1. CRUD 操作

SQLAlchemy 提供了完整的 CRUD（Create, Read, Update, Delete）操作介面。

# 新增資料
new_user = User(name='John Doe', age=30)
session.add(new_user)
session.commit()

# 查詢資料
users = session.query(User).all()

# 更新資料
user = session.query(User).filter_by(name='John Doe').first()
user.age = 31
session.commit()

# 刪除資料
session.delete(user)
session.commit()

內容解密：

1. session.add(new_user)：將新物件加入 Session，準備寫入資料函式庫。
2. session.commit()：提交 Session 中的變更到資料函式庫。
3. session.query(User).all()：查詢 User 表中的所有記錄。
4. session.query(User).filter_by(name='John Doe').first()：根據條件查詢第一筆符合的記錄。
5. session.delete(user)：刪除指定的物件。

2. 查詢方法

• all(): 取得所有查詢結果。
• first(): 取得第一筆查詢結果。
• one(): 確保查詢結果只有一筆，否則丟擲異常。
• one_or_none(): 取得一筆查詢結果，若無結果則傳回 None。

資料函式庫組態與遷移

1. 資料函式庫 URL

SQLAlchemy 使用特定的 URL 格式來連線不同型別的資料函式庫，例如：

# PostgreSQL
DATABASE_URL = "postgresql://user:password@localhost/dbname"

# MySQL
DATABASE_URL = "mysql+pymysql://user:password@localhost/dbname"

# SQLite
DATABASE_URL = "sqlite:///path/to/db.sqlite"

內容解密：

1. postgresql://、mysql+pymysql://、sqlite:///：分別表示連線到 PostgreSQL、MySQL（使用 pymysql 驅動）和 SQLite 資料函式庫。
2. user:password：資料函式庫的使用者名稱和密碼。
3. localhost：資料函式庫伺服器的位址。
4. dbname 或 path/to/db.sqlite：資料函式庫名稱或 SQLite 資料函式庫檔案路徑。

2. Alembic 資料函式庫遷移工具

Alembic 是 SQLAlchemy 官方提供的資料函式庫遷移工具，用於管理資料函式庫結構的變更。

# 初始化 Alembic
alembic init alembic

# 生成遷移指令碼
alembic revision --autogenerate

# 套用遷移
alembic upgrade head

內容解密：

1. alembic init alembic：初始化 Alembic，建立相關的組態檔案和目錄。
2. alembic revision --autogenerate：根據當前 Model 的定義自動生成遷移指令碼。
3. alembic upgrade head：將資料函式庫升級到最新的版本。

關聯與關係載入

1. 關聯型別

SQLAlchemy 支援多種關聯型別，包括一對多、多對一、多對多等。

class Parent(Base):
    __tablename__ = 'parents'
    id = Column(Integer, primary_key=True)
    children = relationship("Child", back_populates="parent")

class Child(Base):
    __tablename__ = 'children'
    id = Column(Integer, primary_key=True)
    parent_id = Column(Integer, ForeignKey('parents.id'))
    parent = relationship("Parent", back_populates="children")

內容解密：

1. relationship("Child", back_populates="parent")：定義 Parent 與 Child 的一對多關聯。
2. ForeignKey('parents.id')：定義 Child 表中的 parent_id 外部索引鍵，參照 Parent 表的 id。

2. 關係載入策略

SQLAlchemy 提供了多種關係載入策略，如懶載入（lazy loading）、急切載入（eager loading）等，以最佳化查詢效能。

# 懶載入（預設）
children = relationship("Child", lazy='selectin')

# 急切載入
children = relationship("Child", lazy='joined')

內容解密：

1. lazy='selectin'：使用 selectin 載入策略，一次查詢載入相關物件。
2. lazy='joined'：使用 joined 載入策略，透過 JOIN 陳述式一次載入相關物件。

非同步操作與擴充套件

1. 非同步 SQLAlchemy

SQLAlchemy 支援非同步操作，透過 async_sessionmaker 和非同步引擎實作。

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

async_engine = create_async_engine("postgresql+asyncpg://user:password@localhost/dbname")
AsyncSession = async_sessionmaker(async_engine, expire_on_commit=False)

async with AsyncSession() as session:
    # 非同步查詢或操作
    pass

內容解密：

1. create_async_engine：建立一個非同步的資料函式庫引擎。
2. async_sessionmaker：建立一個非同步的 Session 工廠。
3. expire_on_commit=False：防止在 commit 後過期物件屬性。