領域模型儲存函式庫模式實踐

在實際應用中，領域模型需要與外部狀態進行互動，例如資料函式庫。為了保持領域模型的純粹性，避免與資料函式庫直接耦合，我們可以使用儲存函式庫模式。儲存函式庫模式作為一個抽象層，隱藏了資料存取的細節，讓領域模型可以專注於業務邏輯。Python 的鴨子型別特性允許我們更靈活地實作儲存函式庫，而 SQLAlchemy 的古典對映則提供了反轉依賴關係的方法，使 ORM 依賴於領域模型，而非相反。透過儲存函式庫模式，我們可以簡化資料存取，提高程式碼的可測試性和可維護性，同時也更容易在不同的儲存方式之間切換。

領域建模與儲存：從模型到儲存函式庫

在上一章中，我們建立了一個簡單的領域模型，能夠將訂單分配到庫存批次。這個模型易於測試，因為它不依賴任何外部基礎設施。然而，當我們需要將這個模型投入實際使用時，就必須考慮如何將其與外部狀態（如資料函式庫或API）連線起來。

實作領域服務

首先，我們來看看如何實作一個領域服務。領域服務是一種封裝了業務邏輯的物件或函式，它代表了業務概念或流程。在我們的例子中，allocate 函式就是一個領域服務，它負責將訂單行分配到合適的批次。

def allocate(line: OrderLine, batches: List[Batch]) -> str:
    try:
        batch = next(b for b in sorted(batches) if b.can_allocate(line))
        batch.allocate(line)
        return batch.reference
    except StopIteration:
        raise OutOfStock(f'Out of stock for sku {line.sku}')

內容解密：

• allocate 函式接受一個 OrderLine 物件和一批 Batch 物件，傳回被分配到的批次參考。
• 它首先對批次進行排序，然後找到第一個能夠滿足訂單行的批次，並將訂單行分配給它。
• 如果沒有任何批次能夠滿足訂單行，則丟擲 OutOfStock 例外。

使用Python的魔法方法

為了使我們的模型能夠與 Python 的內建函式（如 sorted）無縫工作，我們實作了 __gt__ 魔法方法。這使得我們能夠根據批次的 eta 屬性對其進行排序。

class Batch:
    ...
    def __gt__(self, other):
        if self.eta is None:
            return False
        if other.eta is None:
            return True
        return self.eta > other.eta

內容解密：

• __gt__ 方法定義了兩個 Batch 物件之間的比較規則。
• 如果一個批次有 eta 而另一個沒有，則有 eta 的批次被視為更大。
• 如果兩個批次都有 eta，則比較它們的 eta 值。

儲存函式庫模式

為了將我們的領域模型與資料儲存層解耦，我們引入了儲存函式庫模式。儲存函式庫是一種簡化了資料存取的抽象層，使得我們的模型層不再直接依賴於資料函式庫。

圖2-1展示了引入儲存函式庫模式前後的架構變化。

圖示說明

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

title 圖示說明

rectangle "使用" as node1
rectangle "抽象" as node2
rectangle "解耦" as node3

node1 --> node2
node2 --> node3

@enduml

此圖示展示了領域模型如何透過儲存函式庫與資料函式庫進行互動，從而實作瞭解耦。

內容解密：

• 領域模型透過儲存函式庫來存取資料，而不需要直接與資料函式庫互動。
• 儲存函式庫提供了一種抽象，使得資料存取的細節被隱藏起來。

建構API端點與依賴反轉原理的應用

當我們建立第一個API端點時，我們知道我們的程式碼將會類別似於以下的形式。

第一個API端點的雛形

@flask.route.gubbins
def allocate_endpoint():
    # 從請求中提取訂單明細
    line = OrderLine(request.params, ...)
    # 從資料函式庫載入所有批次
    batches = ...
    # 呼叫我們的領域服務
    allocate(line, batches)
    # 然後將分配結果儲存回資料函式庫
    return 201

內容解密：

1. 提取訂單明細：從HTTP請求中取得訂單明細的資訊。
2. 載入批次資料：從資料函式庫中取出相關的批次資料。
3. 呼叫領域服務：使用allocate函式處理訂單明細和批次資料之間的分配邏輯。
4. 儲存結果：將分配結果儲存到資料函式庫中。

依賴反轉原理（DIP）與資料存取

在介紹中提到，分層架構是構建具有UI、邏輯和資料函式庫系統的常見方法（參見圖2-2）。然而，我們希望我們的領域模型完全獨立，不依賴於任何基礎設施。為此，我們採用了洋蔥架構（Onion Architecture），使依賴關係朝向領域模型內部流動（參見圖2-3）。

我們的領域模型

讓我們回顧一下我們的領域模型（參見圖2-4）：分配是將OrderLine連結到Batch的概念。我們將分配儲存為Batch物件上的集合。

將領域模型對映到關聯式資料函式庫

現在，讓我們看看如何將這個模型轉換為關聯式資料函式庫。

常規的ORM方式：模型依賴於ORM

大多數團隊使用物件關聯對映器（ORM）來根據模型物件生成SQL查詢。以SQLAlchemy為例，典型的用法如下：

from sqlalchemy import Column, ForeignKey, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

Base = declarative_base()

class Order(Base):
    id = Column(Integer, primary_key=True)

class OrderLine(Base):
    id = Column(Integer, primary_key=True)
    sku = Column(String(250))
    qty = Column(Integer)  # 更正：Integer而非Integer(String(250))
    order_id = Column(Integer, ForeignKey('order.id'))
    order = relationship(Order)

class Allocation(Base):
    ...

內容解密：

1. 定義模型類別：使用SQLAlchemy的宣告式語法定義Order、OrderLine和Allocation等模型類別。
2. 欄位定義：定義每個模型類別的欄位，如id、sku、qty等。
3. 關聯定義：使用ForeignKey和relationship定義模型之間的關聯。

然而，這種方式使得我們的領域模型直接依賴於ORM框架，違背了永續性無知的原則。

反轉依賴：ORM依賴於模型

幸運的是，SQLAlchemy提供了另一種使用方式，即明確定義schema和對映器（mapper），實作ORM依賴於領域模型。這種方式稱為古典對映（Classical Mapping）。

from sqlalchemy.orm import mapper, relationship
import model
metadata = MetaData()

# 定義表格和對映器
order_lines = Table('order_lines', metadata,
                    Column('id', Integer, primary_key=True),
                    Column('sku', String(250)),
                    Column('qty', Integer),
                    Column('order_id', Integer, ForeignKey('orders.id'))
                   )

mapper(model.OrderLine, order_lines)

內容解密：

1. 定義表格：使用Table物件定義資料函式庫表格結構。
2. 定義對映器：使用mapper函式將領域模型類別（如OrderLine）對映到資料函式庫表格。
3. 實作永續性無知：領域模型不再依賴於特定的ORM框架，實作了永續性無知。

這種方式使得我們的領域模型保持乾淨和獨立，不受基礎設施層的影響。

儲存函式庫模式（Repository Pattern）介紹

儲存函式庫模式是一種對持久化儲存的抽象，透過隱藏資料存取的繁瑣細節，模擬所有資料都存在記憶體中的效果。如果我們的電腦有無限的記憶體，我們就不需要笨重的資料函式庫，可以隨時使用物件。

記憶體中的資料存取

在記憶體中，我們需要將物件存放在某處，以便稍後再次找到它們。我們的記憶體資料結構可以讓我們新增物件，就像使用列表或集合一樣。由於物件都在記憶體中，我們不需要呼叫 .save() 方法，只需擷取關心的物件並在記憶體中修改即可。

抽象儲存函式庫

最簡單的儲存函式庫只有兩個方法：add() 用於將新專案新增至儲存函式庫，get() 用於傳回先前新增的專案。我們嚴格遵守在領域和服務層中使用這些方法進行資料存取，這種自律的簡化避免了將領域模型與資料函式庫耦合。

抽象儲存函式庫類別實作

# repository.py
class AbstractRepository(abc.ABC):
    @abc.abstractmethod
    def add(self, batch: model.Batch):
        raise NotImplementedError

    @abc.abstractmethod
    def get(self, reference) -> model.Batch:
        raise NotImplementedError

程式碼解密：

1. AbstractRepository 類別定義：使用 abc.ABC 作為基礎類別，定義了一個抽象儲存函式庫類別。
2. @abc.abstractmethod 修飾符：標記 add 和 get 方法為抽象方法，強制子類別實作這些方法。
3. add 方法：用於將新的 Batch 物件新增至儲存函式庫中。
4. get 方法：根據參考值（如批次參考編號）傳回對應的 Batch 物件。
5. raise NotImplementedError：在抽象方法中丟擲 NotImplementedError 異常，表示該方法必須由子類別實作。

為何使用儲存函式庫模式

使用儲存函式庫模式可以讓領域模型與基礎設施（如資料函式庫）解耦，提高程式碼的可測試性和可維護性。透過定義一個簡潔的介面，我們可以輕鬆地在測試中偽造（fake）儲存函式庫，從而專注於業務邏輯的測試。

ORM 與領域模型的解耦

在前面的章節中，我們已經透過 SQLAlchemy 的 ORM 功能，將領域模型與資料函式庫表進行了對映。這種做法雖然方便，但仍然存在領域模型與 ORM 框架的耦合。透過引入儲存函式庫模式，我們可以進一步將領域模型與 ORM 解耦，使得領域模型更加純粹，不依賴於任何基礎設施。

範例：測試 ORM 組態

# test_orm.py
def test_orderline_mapper_can_load_lines(session):
    session.execute(
        'INSERT INTO order_lines (orderid, sku, qty) VALUES '
        '("order1", "RED-CHAIR", 12),'
        '("order1", "RED-TABLE", 13),'
        '("order2", "BLUE-LIPSTICK", 14)'
    )
    expected = [
        model.OrderLine("order1", "RED-CHAIR", 12),
        model.OrderLine("order1", "RED-TABLE", 13),
        model.OrderLine("order2", "BLUE-LIPSTICK", 14),
    ]
    assert session.query(model.OrderLine).all() == expected

程式碼解密：

1. test_orderline_mapper_can_load_lines 函式：測試 OrderLine 對映器是否能正確從資料函式庫載入資料。
2. session.execute 方法：直接執行 SQL 陳述式，向 order_lines 表中插入測試資料。
3. expected 列表：定義預期的 OrderLine 物件列表。
4. assert 陳述式：驗證查詢結果是否與預期相符。

抽象基礎類別、鴨子型別與協定

本文使用抽象基礎類別（Abstract Base Classes, ABCs）主要是為了教學目的，期望能夠清晰地闡述儲存函式庫抽象化的介面。

在實際開發中，我們有時會刪除生產程式碼中的 ABCs，因為 Python 的特性使得它們很容易被忽略，最終導致維護困難甚至誤導。實際上，我們通常依賴 Python 的鴨子型別（duck typing）來實作抽象化。對 Python 開發者來說，一個儲存函式庫（repository）就是任何具有 add(thing) 和 get(id) 方法的物件。

另一種可供選擇的是 PEP 544 協定（protocols）。這些協定提供了型別檢查，而不需要繼承，對於偏好「組合優於繼承」的人來說尤其有用。

權衡利弊

正如經濟學家被認為瞭解一切的價格卻不知道其價值，程式設計師往往只看到技術的好處，卻忽略了其背後的取捨。

— Rich Hickey

每當我們在本文中介紹一種架構模式時，都會問：「這樣做的好處是什麼？代價又是什麼？」

通常，至少我們會引入額外的抽象層。儘管我們希望它能降低整體複雜度，但它確實在區域性增加了複雜性，並且在維護和零件數量上都有所成本。

儲存函式庫模式可能是本文中最容易選擇的模式之一，尤其是當你已經在使用領域驅動設計（DDD）和依賴反轉（dependency inversion）的情況下。就我們的程式碼而言，我們只是將 SQLAlchemy 抽象（session.query(Batch)）替換為我們自己設計的另一個抽象（batches_repo.get）。

每次新增一個想要檢索的領域物件時，我們都需要在儲存函式庫類別中編寫幾行程式碼，但作為回報，我們獲得了一個簡單的儲存層抽象，這個抽象由我們控制。儲存函式庫模式使得對儲存方式進行根本性更改變得容易（參見附錄 C），並且正如我們將看到的，它很容易被偽造以進行單元測試。

此外，儲存函式庫模式在 DDD 社群中非常常見，如果你與來自 Java 和 C# 背景的程式設計師合作，他們很可能會認識這種模式。圖 2-5 說明瞭這種模式。

圖 2-5. 儲存函式庫模式

如同往常，我們從測試開始。這可能被歸類別為整合測試，因為我們正在檢查我們的程式碼（儲存函式庫）是否與資料函式庫正確整合；因此，這些測試傾向於將原始 SQL 與對我們自己的程式碼的呼叫和斷言混合在一起。

提示

與之前的 ORM 測試不同，這些測試更適合長期保留在你的程式碼函式庫中，特別是如果你的領域模型的某些部分使得物件關係對映變得非常重要。

儲存函式庫測試：儲存物件

def test_repository_can_save_a_batch(session):
    batch = model.Batch("batch1", "RUSTY-SOAPDISH", 100, eta=None)
    repo = repository.SqlAlchemyRepository(session)
    repo.add(batch)
    session.commit()
    rows = list(session.execute(
        'SELECT reference, sku, _purchased_quantity, eta FROM "batches"'
    ))
    assert rows == [("batch1", "RUSTY-SOAPDISH", 100, None)]

內容解密：

• test_repository_can_save_a_batch 函式測試儲存函式庫是否能夠正確儲存一個批次物件。
• 首先，建立一個 Batch 物件並使用 SqlAlchemyRepository 將其加入儲存函式庫。
• 然後，提交更改到資料函式庫。
• 使用原始 SQL 查詢資料函式庫，以驗證資料是否正確儲存。

儲存函式庫測試：檢索複雜物件

def insert_order_line(session):
    session.execute(
        'INSERT INTO order_lines (orderid, sku, qty)'
        ' VALUES ("order1", "GENERIC-SOFA", 12)'
    )
    [[orderline_id]] = session.execute(
        'SELECT id FROM order_lines WHERE orderid=:orderid AND sku=:sku',
        dict(orderid="order1", sku="GENERIC-SOFA")
    )
    return orderline_id

def test_repository_can_retrieve_a_batch_with_allocations(session):
    orderline_id = insert_order_line(session)
    batch1_id = insert_batch(session, "batch1")
    insert_batch(session, "batch2")
    insert_allocation(session, orderline_id, batch1_id)

    repo = repository.SqlAlchemyRepository(session)
    retrieved = repo.get("batch1")
    expected = model.Batch("batch1", "GENERIC-SOFA", 100, eta=None)
    assert retrieved == expected  # Batch.__eq__ 只比較 reference
    assert retrieved.sku == expected.sku
    assert retrieved._purchased_quantity == expected._purchased_quantity
    assert retrieved._allocations == {
        model.OrderLine("order1", "GENERIC-SOFA", 12),
    }

內容解密：

• test_repository_can_retrieve_a_batch_with_allocations 函式測試儲存函式庫是否能夠正確檢索具有分配的批次物件。
• 首先，使用輔助函式插入訂單明細和批次資料到資料函式庫。
• 然後，使用 SqlAlchemyRepository 從儲存函式庫中檢索批次物件。
• 最後，驗證檢索到的批次物件的屬性和分配是否正確。

典型的儲存函式庫實作

class SqlAlchemyRepository(AbstractRepository):
    def __init__(self, session):
        self.session = session

    def add(self, batch):
        # 將批次物件加入工作階段
        self.session.add(batch)

內容解密：

• SqlAlchemyRepository 類別是使用 SQLAlchemy 的儲存函式庫實作。
• 建構函式接受一個 session 物件，該物件用於與資料函式庫互動。
• add 方法將批次物件加入工作階段，以便稍後提交到資料函式庫。