SQLAlchemy 銷售資料查詢最佳實踐

– 8 分鐘閱讀

SQLAlchemy 提供了靈活且強大的 ORM 功能，能有效簡化資料函式庫操作。本文將探討如何運用 SQLAlchemy 進行銷售資料的查詢與分析，包含計算訂單總額、篩選特定日期區間的銷售資料、找出熱銷商品等常見應用場景。同時，我們也會示範如何使用 SQLAlchemy 建立多對多關聯模型，例如產品與客戶評論之間的關係，並講解如何處理不同資料函式庫引擎（如 SQLite、MySQL、PostgreSQL）在結果格式上的差異。

透過一系列程式碼範例與解說，讀者將學習如何運用 SQLAlchemy 的核心功能，例如 select、join、group_by、order_by、func.sum、func.avg、func.extract 等，以及如何結合日期時間函式和篩選條件進行更精確的資料查詢。此外，我們也將示範如何使用關聯物件模式建立多對多關係，並透過資料函式庫遷移和資料匯入來完善資料函式庫結構。最後，文章提供一系列進階練習題，涵蓋更複雜的查詢情境，例如多條件篩選、聚合函式應用、多表關聯查詢等，幫助讀者鞏固所學知識並提升實務操作能力，進而應用於真實世界的銷售資料分析專案。

資料函式庫查詢最佳實踐：銷售資料分析

在進行銷售資料分析時，資料函式庫查詢的效率和準確性至關重要。本文將介紹如何使用SQLAlchemy進行複雜的銷售資料查詢，包括計算訂單總金額、找出最暢銷的產品以及在特定日期範圍內進行銷售資料分析。

計算訂單總金額

要計算訂單的總金額，需要將訂單中的每個商品的單價乘以數量，然後將這些金額加總。以下是一個使用SQLAlchemy實作此功能的查詢範例：

order_total = func.sum(OrderItem.unit_price * OrderItem.quantity).label(None)
q = (select(Order, order_total)
     .join(Order.order_items)
     .group_by(Order)
     .order_by(order_total.desc())
     .limit(3))
session.execute(q).all()

內容解密：

  1. order_total：計算每個訂單專案的總金額，並將結果標記為None，表示不給予特定標籤。
  2. select(Order, order_total)：選擇要檢索的欄位，包括Order物件和計算出的order_total。
  3. .join(Order.order_items)：將Order與其相關的OrderItem進行連線，以便計算訂單總金額。
  4. .group_by(Order)：按訂單分組，以便對每個訂單的專案進行匯總。
  5. .order_by(order_total.desc())：按訂單總金額降序排序。
  6. .limit(3)：限制結果只傳回前3個訂單。

找出最暢銷的產品

要找出最暢銷的產品，需要計算每個產品的銷售數量。以下是一個使用SQLAlchemy實作此功能的查詢範例：

units = func.sum(OrderItem.quantity).label(None)
q = (select(Product, units)
     .join(Product.order_items)
     .group_by(Product)
     .order_by(units.desc())
     .limit(5))
session.execute(q).all()

內容解密：

  1. units：計算每個產品的銷售數量總和。
  2. select(Product, units)：選擇要檢索的欄位，包括Product物件和計算出的銷售數量。
  3. .join(Product.order_items)：將Product與其相關的OrderItem進行連線，以便計算銷售數量。
  4. .group_by(Product)：按產品分組，以便對每個產品的銷售數量進行匯總。
  5. .order_by(units.desc())：按銷售數量降序排序。
  6. .limit(5)：限制結果只傳回前5個最暢銷的產品。

在特定日期範圍內進行銷售資料分析

要在特定日期範圍內進行銷售資料分析，需要在查詢中加入日期篩選條件。以下是一個使用SQLAlchemy實作此功能的查詢範例：

from datetime import datetime
order_total = func.sum(OrderItem.unit_price * OrderItem.quantity).label(None)
q = (select(Order, order_total)
     .join(Order.order_items)
     .where(Order.timestamp.between(datetime(2022, 11, 1), datetime(2022, 12, 1)))
     .group_by(Order)
     .order_by(order_total.desc())
     .limit(3))
session.execute(q).all()

內容解密：

  1. where(Order.timestamp.between(datetime(2022, 11, 1), datetime(2022, 12, 1)))：篩選出在2022年11月1日至2022年12月1日之間的訂單。
  2. 其他部分與計算訂單總金額的查詢相同。

結合多表查詢

在某些情況下，需要結合多個表格進行查詢。例如，要在特定日期範圍內找出最暢銷的產品，需要結合Product、OrderItem和Order三個表格。以下是一個使用SQLAlchemy實作此功能的查詢範例：

units = func.sum(OrderItem.quantity).label(None)
q = (select(Product, units)
     .join(Product.order_items)
     .join(OrderItem.order)
     .where(Order.timestamp.between(datetime(2022, 11, 1), datetime(2022, 12, 1)))
     .group_by(Product)
     .order_by(units.desc())
     .limit(5))
session.execute(q).all()

內容解密：

  1. .join(OrderItem.order)：將OrderItem與其相關的Order進行連線，以便篩選出特定日期範圍內的訂單。
  2. where(Order.timestamp.between(datetime(2022, 11, 1), datetime(2022, 12, 1)))：篩選出在2022年11月1日至2022年12月1日之間的訂單。
  3. 其他部分與找出最暢銷的產品的查詢相同。

透過上述範例，可以看出使用SQLAlchemy可以輕鬆地進行複雜的銷售資料查詢和分析。只要掌握了基本的查詢技巧和函式使用，就可以根據不同的業務需求進行靈活的資料分析。

資料函式庫查詢與多對多關係的進階應用

在處理複雜的資料函式庫查詢時，SQLAlchemy 提供了豐富的功能來滿足各種需求。上一節討論瞭如何使用 between() 篩選條件來限制訂單時間戳的結果範圍。本文將探討另一種常見的查詢模式：按時間單位（如天、月、季度或年）對結果進行分組。

按時間單位分組查詢結果

要實作按時間單位分組，需要將時間戳轉換為可以在 group_by() 子句中使用的形式。這樣，所有屬於同一時間間隔的結果就可以被聚合起來。

以下查詢使用 extract() 函式從訂單時間戳中提取年份和月份，然後按這些值進行分組，以計算 2022 年每月的總銷售量。

month = func.extract('month', Order.timestamp).label(None)
year = func.extract('year', Order.timestamp).label(None)
units = func.sum(OrderItem.quantity).label(None)
q = (select(year, month, units)
     .join(OrderItem)
     .where(Order.timestamp.between(datetime(2022, 1, 1), datetime(2023, 1, 1)))
     .group_by(year, month)
     .order_by(year, month))
result = session.execute(q).all()
print(result)  # [(2022, 1, 505), (2022, 2, 426), (2022, 3, 525), ..., (2022, 12, 564)]

內容解密：

  1. func.extract()：此函式用於從日期時間欄位中提取特定的部分，如年、月、日等。
    • 第一個引數指定要提取的部分（例如 ‘year’、‘month’）。
    • 第二個引數是日期時間欄位（例如 Order.timestamp）。
  2. label(None)：為提取的值或聚合結果指定標籤。在這裡，None 表示不使用特定的標籤名稱。
  3. group_by(year, month)：根據提取的年份和月份對結果進行分組。
  4. order_by(year, month)：按年份和月份對結果進行排序。

處理不同資料函式庫引擎的結果差異

不同的資料函式庫引擎可能會以不同的格式傳回計算結果或函式結果。例如，使用 SQLite 時，結果是標準的整數和浮點數；使用 MySQL 時，總和是 Decimal 物件；使用 PostgreSQL 時，提取的值是 Decimal 物件，而總和是整數。

# MySQL 結果示例
[(2022, 1, Decimal('505')), (2022, 2, Decimal('426')), ..., (2022, 12, Decimal('564'))]

# PostgreSQL 結果示例
[(Decimal('2022'), Decimal('1'), 505), (Decimal('2022'), Decimal('2'), 426), ..., (Decimal('2022'), Decimal('12'), 564)]

要處理 Decimal 物件，可以使用 int() 或 float() 函式進行轉換。

from decimal import Decimal
int(Decimal('2022'))  # 輸出：2022
float(Decimal('2022'))  # 輸出：2022.0

新增客戶評論功能與多對多關係

為了支援客戶評論，需要在產品和客戶之間建立新的多對多關係。由於評論需要額外的儲存空間來存放評分和評論文字，因此採用根據關聯物件模式的高階解決方案。

class ProductReview(Model):
    __tablename__ = 'products_reviews'
    product_id: Mapped[int] = mapped_column(ForeignKey('products.id'), primary_key=True)
    customer_id: Mapped[UUID] = mapped_column(ForeignKey('customers.id'), primary_key=True)
    timestamp: Mapped[datetime] = mapped_column(default=datetime.utcnow, index=True)
    rating: Mapped[int]
    comment: Mapped[Optional[str]] = mapped_column(Text)

    product: Mapped['Product'] = relationship(back_populates='reviews')
    customer: Mapped['Customer'] = relationship(back_populates='product_reviews')

內容解密：

  1. ProductReview 模型：代表產品評論的關聯表，包含產品 ID、客戶 ID、時間戳、評分和評論文字等欄位。
    • product_id 和 customer_id 是外部索引鍵，分別指向 products 和 customers 表的主鍵。
    • timestamp 自動預設為目前時間，並建立索引以便快速查詢。
    • rating 是客戶給出的評分。
    • comment 是客戶的評論文字，使用 Text 型別來儲存較長的文字。
  2. relationship：定義了 ProductReview 與 Product 和 Customer 之間的關係。
    • back_populates 引數確保雙向關係的一致性。

資料函式庫遷移與資料匯入

新增模型後，需要生成資料函式庫遷移指令碼並應用到資料函式庫。

(venv) $ alembic revision --autogenerate -m "product reviews"
(venv) $ alembic upgrade head

為了測試查詢，可以匯入一批產品評論資料。

# import_reviews.py
import csv
from datetime import datetime
from sqlalchemy import select, delete
# ... 資料匯入邏輯 ...

內容解密：

  1. 匯入 CSV 資料：使用 Python 的 csv 模組讀取 CSV 檔案，並將資料匯入到資料函式庫中。
  2. select 和 delete：使用 SQLAlchemy 的查詢語法來檢查或刪除現有資料。

資料函式庫查詢與分析實務

在前面的章節中，我們已經建立了一個完整的資料函式庫模型並匯入了相關的資料。現在，我們將透過一系列的查詢和分析來展示如何有效地使用SQLAlchemy進行資料檢索和處理。

查詢例項

首先，讓我們開始一個Python shell並匯入必要的函式和類別，然後建立一個資料函式庫會話（Session）：

>>> from sqlalchemy import select, func
>>> from db import Session
>>> from models import Product, Customer, ProductReview
>>> session = Session()

計算所有客戶評分的平均值

>>> q = select(func.avg(ProductReview.rating))
>>> session.scalar(q)
3.7731384829505914

這個查詢使用了avg()聚合函式來計算所有評分的平均值。

計算特定產品的平均評分

>>> p = session.scalar(select(Product).where(Product.name == 'ZX Spectrum'))
>>> q = select(func.avg(ProductReview.rating)).where(ProductReview.product == p)
>>> session.scalar(q)
4.0

這裡，我們使用了ProductReview.product關係來建立查詢條件。

生成所有產品的平均評分報告

>>> product_rating = func.avg(ProductReview.rating).label('average_rating')
>>> q = (select(Product, product_rating)
.join(Product.reviews)
.group_by(Product)
.order_by(product_rating.desc(), Product.name))
>>> session.execute(q).all()
[(Product(19, "Apple IIc Plus"), 5.0), ..., (Product(138, "Timex Sinclair 1000"), 1.0)]

這個查詢使用了avg()函式並結合分組，來計算每個產品的平均評分。

計算產品評論中沒有文字的比例

>>> no_comment_percent = (
100 - 100 * func.count(ProductReview.comment) / func.count(ProductReview.rating)
).label('no_comment_percentage')
>>> q = (select(Product.name, no_comment_percent)
.join(ProductReview.product)
.group_by(Product)
.order_by(no_comment_percent.desc(), Product.name))
>>> session.execute(q).all()
[('464 Plus', 100.0), ('Acorn Atom', 100.0), ..., ('ZX81', 0.0)]

這個查詢計算了每個產品中沒有文字評論的比例。

練習題

現在是時候自己動手練習了。請寫出以下查詢：

  1. 超過300美元的訂單，按銷售金額從高到低排序。
  2. 包含一台或多台ZX81電腦的訂單。
  3. 包含Amstrad製造的產品的訂單。
  4. 2022年12月25日下的訂單，且包含兩個或以上的訂單專案。
  5. 客戶的第一筆和最後一筆訂單日期和時間。
  6. 銷售金額最高的Top 5製造商，按銷售金額從高到低排序。
  7. 產品、平均星級評分和評論數量，按評論數量從高到低排序。
  8. 產品和平均星級評分，但只計算包含文字評論的評分。
  9. 2022年每個月份Commodore 64電腦的平均星級評分。
  10. 客戶給予產品的最低和最高星級評分，按客戶名稱字母順序排序。
  11. 製造商的平均星級評分，按平均評分從高到低排序。
  12. 產品國家的平均星級評分，按平均評分從高到低排序。

網頁流量分析解決方案

本章的目標是使用您所學的概念來建立一個網頁流量分析解決方案。這將作為前面章節中演示的技術的強化，以及一個更複雜和真實的資料函式庫設計範例。

玄貓 BlackCat

玄貓 BlackCat

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。