在建構電子商務平台或類別似系統時，有效管理客戶和訂單資料至關重要。本文將詳細介紹如何運用 SQLAlchemy，一個強大的 Python ORM 框架，來設計和實作一個穩健的客戶訂單管理系統。我們將探討資料函式庫模型的設計，特別是如何處理多對多關係以及使用 write_only 最佳化查詢效能。同時，也會涵蓋資料函式庫遷移、訂單建立流程、以及如何利用 CSV 檔案匯入大量訂單資料等實務技巧。
建立訂單與客戶管理系統的資料函式庫模型
在前一章的基礎上，本章將探討如何使用 SQLAlchemy 實作客戶與訂單管理系統的資料函式庫模型。主要內容包括如何使用 write_only 載入器來處理大型資料集，以及如何使用 Association Object Pattern 來建立具有額外欄位的多對多關係。
使用 write_only 載入器處理大型資料集
在處理大型資料集時，直接載入所有相關資料可能會導致效能問題。write_only 載入器提供了一個解決方案，它允許開發者延遲載入相關資料，直到真正需要時才執行查詢。這種方式特別適合於處理大量資料的情況。
class Product(Model):
# ...
order_items: WriteOnlyMapped['OrderItem'] = relationship(
back_populates='product')
# ...
#### 內容解密：
- `WriteOnlyMapped` 用於定義一個 `write_only` 的關係屬性。
- `relationship` 函式用於建立模型之間的關係。
- `back_populates` 引數用於指定對應的反向關係屬性名稱。
### 使用 Association Object Pattern 建立多對多關係
當需要在多對多關係中新增額外的欄位時，簡單的多對多關係就無法滿足需求。這時，可以使用 Association Object Pattern 來建立具有額外欄位的關聯表。
#### OrderItem 模型的定義
```python
class OrderItem(Model):
__tablename__ = 'orders_items'
product_id: Mapped[int] = mapped_column(ForeignKey('products.id'), primary_key=True)
order_id: Mapped[UUID] = mapped_column(ForeignKey('orders.id'), primary_key=True)
unit_price: Mapped[float]
quantity: Mapped[int]
#### 內容解密：
- `OrderItem` 模型代表訂單中的每一項產品。
- `product_id` 和 `order_id` 是外部索引鍵，分別對應到 `Product` 和 `Order` 模型的主鍵。
- `unit_price` 和 `quantity` 用於儲存訂單中每項產品的單價和數量。
### 分解多對多關係
為了實作具有額外欄位的多對多關係，可以將其分解為兩個一對多的關係。
```python
class Product(Model):
# ...
order_items: WriteOnlyMapped['OrderItem'] = relationship(
back_populates='product')
# ...
class Order(Model):
# ...
order_items: Mapped[list['OrderItem']] = relationship(
back_populates='order')
# ...
class OrderItem(Model):
# ...
product: Mapped['Product'] = relationship(back_populates='order_items')
order: Mapped['Order'] = relationship(back_populates='order_items')
# ...
#### 內容解密：
- `Product` 和 `Order` 模型透過 `OrderItem` 模型建立多對多關係。
- `order_items` 關係屬性允許存取訂單中的所有專案或產品的所有訂單記錄。
- 使用 `write_only` 載入器來處理可能的大型資料集。
### 資料函式庫遷移
完成模型定義後，需要進行資料函式庫遷移以更新資料函式庫結構。
```bash
(venv) $ alembic revision --autogenerate -m "customers and orders"
(venv) $ alembic upgrade head
內容解密：
- 使用
alembic revision命令自動生成遷移指令碼。 - 使用
alembic upgrade命令將遷移指令碼應用到資料函式庫中。
建立訂單的步驟
建立具有額外欄位的多對多關係需要更多的管理工作，但提供了更大的靈活性。以下是建立訂單的步驟：
- 建立新的
Order例項。 - 為訂單新增產品，建立相應的
OrderItem例項。 - 設定每個
OrderItem的unit_price和quantity。
透過這些步驟，可以有效地管理客戶訂單和相關產品資訊。
建立客戶訂單的流程與實作細節
在電子商務系統中，建立客戶訂單是一個關鍵的業務流程。本文將詳細介紹如何使用 SQLAlchemy 這個流行的 Python SQL 工具包和物件關係對映（ORM）函式庫來建立客戶訂單。
客戶與訂單的關聯
首先，需要建立一個客戶（Customer）例項，如果是重複客戶，則載入現有的客戶記錄。接著，建立一個訂單（Order）例項，並將其與客戶關聯起來。這可以透過在建構函式中傳遞客戶引數或呼叫 Customer.orders 關係上的 add() 方法來實作。
# 建立新客戶
c = Customer(name='Jane Smith')
# 建立新訂單，並將其新增到客戶和資料函式庫會話中
o = Order()
c.orders.add(o)
session.add(o)
新增訂單專案
對於訂單中的每個明細專案，需要建立一個 OrderItem 例項，包含產品、單價和數量，並將其附加到訂單的 order_items 關係中。
# 新增第一個訂單專案：產品 #45，價格 $45.50
p1 = session.get(Product, 45)
o.order_items.append(OrderItem(product=p1, unit_price=45.5, quantity=1))
# 新增第二個訂單專案：2 個產品 #82，每個 $37
p2 = session.get(Product, 82)
o.order_items.append(OrderItem(product=p2, unit_price=37, quantity=2))
內容解密：
session.get(Product, 45)：從資料函式庫會話中根據 ID 取得產品例項。o.order_items.append(OrderItem(...))：建立OrderItem例項並將其附加到訂單的order_items關係中，實作訂單與產品的多對多關聯。unit_price和quantity：記錄訂單專案的單價和數量，用於計算總價。
處理刪除操作
當刪除一個被多個 OrderItem 參照的產品或訂單時，SQLAlchemy 會嘗試將無效的外部索引鍵寫入 OrderItem 例項。由於外部索引鍵被定義為非空，因此這種嘗試將失敗。有兩種方法可以解決這個問題：一是組態級聯刪除，二是假設當存在參照時，相關實體不能被刪除。
# 從訂單中刪除 OrderItem
o.order_items.remove(oi)
session.commit()
# 從產品中刪除 OrderItem
p.order_items.delete(oi)
session.commit()
內容解密：
o.order_items.remove(oi)：從訂單的order_items關係中移除指定的OrderItem例項。p.order_items.delete(oi)：從產品的order_items關係中刪除指定的OrderItem例項。session.commit()：提交變更到資料函式庫。
匯入訂單指令碼
為了方便測試和實驗，可以使用一個指令碼從 CSV 檔案匯入大量隨機生成的訂單。
import csv
from datetime import datetime
from sqlalchemy import select, delete
from db import Session
from models import Product, Customer, Order, OrderItem
def main():
with Session() as session:
with session.begin():
# 清除現有資料
session.execute(delete(OrderItem))
session.execute(delete(Order))
session.execute(delete(Customer))
# 從 CSV 檔案匯入訂單
with open('orders.csv') as f:
reader = csv.DictReader(f)
all_customers = {}
all_products = {}
for row in reader:
# 建立客戶和訂單
if row['name'] not in all_customers:
c = Customer(name=row['name'], address=row['address'], phone=row['phone'])
all_customers[row['name']] = c
o = Order(timestamp=datetime.strptime(row['timestamp'], '%Y-%m-%d %H:%M:%S'))
all_customers[row['name']].orders.add(o)
session.add(o)
# 建立訂單專案
product = all_products.get(row['product1'])
if product is None:
product = session.scalar(select(Product).where(Product.name == row['product1']))
all_products[row['product1']] = product
o.order_items.append(OrderItem(product=product, unit_price=float(row['unit_price1']), quantity=int(row['quantity1'])))
內容解密：
csv.DictReader(f)：讀取 CSV 檔案並將每一行轉換為字典，方便存取欄位值。all_customers和all_products：快取客戶和產品例項，避免重複查詢資料函式庫。session.scalar(select(Product).where(Product.name == row['product1']))：根據產品名稱查詢產品例項。o.order_items.append(OrderItem(...))：建立OrderItem例項並將其附加到訂單的order_items關係中。
綜上所述，使用 SQLAlchemy 可以有效地建立和管理客戶訂單，包括新增、刪除和匯入訂單等操作。透過合理組態關係和級聯操作，可以確保資料的一致性和完整性。
訂單匯入指令碼與查詢操作解析
訂單資料匯入指令碼詳解
該指令碼主要功能為從CSV檔案中讀取訂單資料，並將其匯入至資料函式庫中。指令碼首先清除現有的訂單和客戶資料，以確保資料函式庫的乾淨狀態。
資料處理邏輯
- 客戶資料處理：指令碼透過
all_customers字典快取已建立的客戶例項，以避免重複查詢或建立。 - 產品資料處理：使用
all_products字典快取已查詢的產品例項，降低對資料函式庫的查詢次數。 - 訂單建立：根據CSV中的每一行資料，建立對應的
Order例項，並將其與客戶進行關聯。同時，設定timestamp屬性為CSV檔案中提供的日期和時間。 - 訂單專案建立：根據CSV檔案中的產品資訊（最多三個產品），建立對應的
OrderItem例項，並將其附加至Order例項。
程式碼重點解析
if row['product1']:
product = all_products.get(row['product1'])
if product is None:
product = session.scalar(select(Product).where(Product.name == row['product1']))
all_products[row['product1']] = product
o.order_items.append(OrderItem(
product=product,
unit_price=float(row['unit_price1']),
quantity=int(row['quantity1'])))
內容解密：
- 首先檢查CSV行中是否包含
product1的資料，若存在則進行處理。 - 嘗試從
all_products快取字典中取得對應的產品例項。若不存在，則透過資料函式庫查詢取得，並將其加入快取。 - 建立
OrderItem例項，將產品、單價和數量資訊填入，並將其附加至訂單。
查詢操作範例
在完成訂單資料匯入後，可利用SQLAlchemy進行多樣化的查詢操作。
使用write_only關係進行查詢
c = session.scalar(select(Customer).where(Customer.name == 'John Butler'))
orders = session.scalars(c.orders.select()).all()
內容解密：
- 首先透過客戶名稱查詢特定的客戶例項。
- 利用
write_only關係屬性取得對應的訂單查詢物件，並執行查詢以取得所有相關訂單。
進階查詢範例
# 排序訂單（由新至舊）
orders = session.scalars(c.orders.select().order_by(Order.timestamp.desc())).all()
# 取得最多一筆訂單
order = session.scalar(c.orders.select().limit(1))
內容解密：
- 示範如何利用SQLAlchemy的查詢介面進行排序和限制結果數量的操作。
- 這些查詢操作充分展現了
write_only關係的彈性。
統計查詢範例
customer_count = session.scalar(select(func.count(Customer.id)))
order_count = session.scalar(select(func.count(Order.id)))
內容解密：
- 利用SQLAlchemy的
func.count功能統計客戶和訂單的總數。 - 這類別統計查詢對於瞭解系統中的資料量有很大幫助。
