返回文章列表
玄貓（BlackCat）

依賴注入與引導指令碼解耦應用

本文探討如何使用 Python

軟體設計 Python
依賴注入 引導指令碼 訊息匯流排 單元測試 整合測試 Python

依賴注入是一種設計模式，旨在降低程式碼耦合度，提升應用程式的可維護性和可測試性。本文探討如何結合引導指令碼和手動依賴注入，建構更具彈性的應用程式架構。引導指令碼負責初始化核心物件和注入依賴，而手動注入則確保依賴關係明確且易於管理。文章將進一步說明如何利用 Python 的特性（如閉包、偏函式和類別）實作手動依賴注入，並以訊息匯流排的設計為例，展示如何將其應用於實際專案中。最後，文章還會探討如何在單元測試和整合測試中，透過自定義訊息匯流排和模擬物件，有效地驗證程式碼的正確性，確保應用程式的穩定性和可靠性。

依賴注入與引導指令碼：解耦應用程式的關鍵

在軟體開發中，管理依賴關係是一項重要的任務。適當的依賴注入（Dependency Injection, DI）能夠提高程式碼的可測試性、可維護性和擴充套件性。本文將探討如何使用引導指令碼（Bootstrap Script）和手動依賴注入來最佳化應用程式的架構。

顯式依賴 vs. 隱式依賴

首先，我們需要了解顯式依賴和隱式依賴之間的區別。顯式依賴是指在函式或類別的簽名中明確宣告的依賴，而隱式依賴則是透過其他方式（如全域性變數或模組匯入）引入的依賴。

# 隱式依賴
from email import send_mail
def send_out_of_stock_notification(event: events.OutOfStock):
    send_mail('[email protected]', f'Out of stock for {event.sku}')

# 顯式依賴
def send_out_of_stock_notification(event: events.OutOfStock, send_mail: Callable):
    send_mail('[email protected]', f'Out of stock for {event.sku}')

內容解密：

  1. 顯式依賴的優點：使測試更容易編寫和管理，因為可以輕易地替換依賴。
  2. 符合依賴倒置原則：依賴於抽象而不是具體實作，提高了程式碼的靈活性。

手動依賴注入

手動依賴注入是一種不使用框架，直接在程式碼中管理依賴關係的方法。它可以透過閉包（Closures）、偏函式（Partial Functions）或類別來實作。

使用閉包或偏函式

def allocate(cmd: commands.Allocate, uow: unit_of_work.AbstractUnitOfWork):
    # 原始的allocate函式
    line = OrderLine(cmd.orderid, cmd.sku, cmd.qty)
    with uow:
        # 處理邏輯
        pass

# 使用閉包
def bootstrap():
    uow = unit_of_work.SqlAlchemyUnitOfWork()
    allocate_composed = lambda cmd: allocate(cmd, uow)
    return allocate_composed

# 使用偏函式
import functools
allocate_composed = functools.partial(allocate, uow=uow)

內容解密：

  1. 閉包和偏函式的作用：將依賴注入到函式中，使其變成一個可直接呼叫的函式。
  2. 區別：閉包使用晚繫結（Late Binding），可能導致可變依賴的問題。

使用類別

class AllocateHandler:
    def __init__(self, uow: unit_of_work.AbstractUnitOfWork):
        self.uow = uow

    def __call__(self, cmd: commands.Allocate):
        line = OrderLine(cmd.orderid, cmd.sku, cmd.qty)
        with self.uow:
            # 處理邏輯
            pass

# 初始化
uow = unit_of_work.SqlAlchemyUnitOfWork()
allocate = AllocateHandler(uow)

內容解密：

  1. 類別的作用：透過__init__宣告依賴，並透過__call__使例項可呼叫。
  2. 適用場景：適合於需要明確依賴關係和複雜初始化邏輯的場景。

引導指令碼

引導指令碼負責初始化應用程式的核心物件，如訊息匯流排（Message Bus），並注入必要的依賴。

def bootstrap(
    start_orm: bool = True,
    uow: unit_of_work.AbstractUnitOfWork = unit_of_work.SqlAlchemyUnitOfWork(),
    send_mail: Callable = email.send,
    publish: Callable = redis_eventpublisher.publish,
) -> messagebus.MessageBus:
    if start_orm:
        orm.start_mappers()
    
    dependencies = {'uow': uow, 'send_mail': send_mail, 'publish': publish}
    injected_event_handlers = {
        event_type: [
            inject_dependencies(handler, dependencies)
            for handler in event_handlers
        ]
        for event_type, event_handlers in handlers.EVENT_HANDLERS.items()
    }
    injected_command_handlers = {
        command_type: inject_dependencies(handler, dependencies)
        for command_type, handler in handlers.COMMAND_HANDLERS.items()
    }
    
    return messagebus.MessageBus(
        uow=uow,
        event_handlers=injected_event_handlers,
        command_handlers=injected_command_handlers,
    )

內容解密：

  1. 引導指令碼的功能：宣告預設依賴、執行初始化工作、注入依賴到處理器。
  2. 靈活性：允許透過引數覆寫預設依賴，以適應不同環境和測試需求。

依賴注入與訊息匯流排的實作

在現代軟體開發中，依賴注入（Dependency Injection, DI）是一種重要的設計模式，用於提高程式碼的模組化、可測試性和可維護性。本文將探討如何在 Python 專案中實作依賴注入，以及如何將其應用於訊息匯流排（Message Bus）的設計中。

透過檢查函式簽名進行依賴注入

首先，我們介紹如何透過檢查函式簽名來實作依賴注入。這種方法涉及使用 Python 的 inspect 模組來檢查處理函式（handler function）的引數，並將相應的依賴注入到這些函式中。

程式碼範例：依賴注入實作

def inject_dependencies(handler, dependencies):
    params = inspect.signature(handler).parameters
    deps = {
        name: dependency
        for name, dependency in dependencies.items()
        if name in params
    }
    return lambda message: handler(message, **deps)

內容解密：

  1. inspect.signature(handler).parameters：檢查 handler 函式的引數簽名，取得其引數名稱和相關資訊。
  2. deps 字典生成式：根據 dependencies 字典和 handler 函式的引數簽名，生成一個新的字典 deps，其中包含需要注入的依賴。
  3. lambda message: handler(message, **deps)：傳回一個新的匿名函式，該函式呼叫原始的 handler 函式，並將 deps 中的依賴以關鍵字引數的形式傳遞給 handler。

手動依賴注入

除了使用 inspect 模組進行自動化的依賴注入外，也可以手動建立部分函式（partial function）來實作依賴注入。這種方法雖然更為繁瑣，但在某些情況下可能更為直觀和易於理解。

程式碼範例：手動建立部分函式

injected_event_handlers = {
    events.Allocated: [
        lambda e: handlers.publish_allocated_event(e, publish),
        lambda e: handlers.add_allocation_to_read_model(e, uow),
    ],
    # 其他事件處理函式...
}

injected_command_handlers = {
    commands.Allocate: lambda c: handlers.allocate(c, uow),
    commands.CreateBatch: lambda c: handlers.add_batch(c, uow),
    # 其他命令處理函式...
}

內容解密：

  1. injected_event_handlers 和 injected_command_handlers 字典：手動建立事件和命令的處理函式，並將依賴（如 uow 和 publish）直接注入到這些處理函式中。
  2. lambda 函式：用於建立部分函式，將特定的依賴注入到處理函式中。

將訊息匯流排設計為可組態的類別

為了使訊息匯流排更加靈活和可組態，我們將其設計為一個類別，並將已經注入依賴的處理函式傳遞給它。

程式碼範例：訊息匯流排類別

class MessageBus:
    def __init__(self, uow, event_handlers, command_handlers):
        self.uow = uow
        self.event_handlers = event_handlers
        self.command_handlers = command_handlers

    def handle(self, message):
        self.queue = [message]
        while self.queue:
            message = self.queue.pop(0)
            if isinstance(message, events.Event):
                self.handle_event(message)
            elif isinstance(message, commands.Command):
                self.handle_command(message)
            else:
                raise Exception(f'{message} was not an Event or Command')

內容解密：

  1. __init__ 方法：初始化訊息匯流排例項，接收工作單元（uow）、事件處理函式和命令處理函式作為引數。
  2. handle 方法：處理傳入的訊息（事件或命令），並根據訊息型別呼叫相應的處理函式。

在應用程式入口點使用載入程式

在應用程式的入口點（如 Flask 路由），我們呼叫載入程式（bootstrap.bootstrap()）來取得一個已組態好的訊息匯流排例項。

程式碼範例：Flask 路由呼叫載入程式

bus = bootstrap.bootstrap()

@app.route("/add_batch", methods=['POST'])
def add_batch():
    cmd = commands.CreateBatch(
        request.json['ref'], request.json['sku'], request.json['qty'], eta,
    )
    bus.handle(cmd)
    return 'OK', 201

內容解密：

  1. bootstrap.bootstrap()：呼叫載入程式，取得一個已組態好的訊息匯流排例項。
  2. bus.handle(cmd)：使用訊息匯流排例項處理傳入的命令。

測試中的自定義訊息匯流排實作

在測試過程中，我們可以利用 bootstrap.bootstrap() 函式並覆寫預設值來取得自定義的訊息匯流排。以下是一個整合測試的範例：

覆寫 Bootstrap 預設值（tests/integration/test_views.py）

@pytest.fixture
def sqlite_bus(sqlite_session_factory):
    bus = bootstrap.bootstrap(
        start_orm=True,
        uow=unit_of_work.SqlAlchemyUnitOfWork(sqlite_session_factory),
        send_mail=lambda *args: None,
        publish=lambda *args: None,
    )
    yield bus
    clear_mappers()

def test_allocations_view(sqlite_bus):
    sqlite_bus.handle(commands.CreateBatch('sku1batch', 'sku1', 50, None))
    sqlite_bus.handle(commands.CreateBatch('sku2batch', 'sku2', 50, date.today()))
    # ...
    assert views.allocations('order1', sqlite_bus.uow) == [
        {'sku': 'sku1', 'batchref': 'sku1batch'},
        {'sku': 'sku2', 'batchref': 'sku2batch'},
    ]

內容解密：

  1. sqlite_bus fixture 的建立：使用 bootstrap.bootstrap() 建立一個自定義的訊息匯流排，其中 start_orm=True 表示啟動 ORM，uow 使用 SqlAlchemyUnitOfWork 並傳入 sqlite_session_factory 以建立資料函式庫連線。
  2. send_mail 和 publish 的處理：將 send_mail 和 publish 設定為 lambda *args: None，表示在測試中不實際傳送郵件或發布訊息，以避免副作用。
  3. test_allocations_view 測試邏輯：透過 sqlite_bus.handle() 處理命令，建立批次並驗證分配結果。

在單元測試中使用 FakeUnitOfWork

在單元測試中，我們可以使用 FakeUnitOfWork 來簡化測試：

def bootstrap_test_app():
    return bootstrap.bootstrap(
        start_orm=False,
        uow=FakeUnitOfWork(),
        send_mail=lambda *args: None,
        publish=lambda *args: None,
    )

內容解密：

  1. start_orm=False：由於使用 FakeUnitOfWork，因此不需要啟動 ORM。
  2. FakeUnitOfWork 的使用：簡化單元測試中的工作單元實作，避免實際的資料函式庫操作。