Repository 模式提供了一種有效的方法來管理資料存取邏輯，並將其與業務邏輯分離。在 Python 中，可以使用 SQLAlchemy 等 ORM 工具來實作 Repository，也可以不使用 ORM 直接操作資料函式庫。Fake Repository 的使用簡化了單元測試的流程。更進一步，可以結合埠和介面卡模式，讓程式碼更具彈性。但是，Repository 模式也存在一些缺點，例如增加了程式碼的複雜度和維護成本，需要根據專案的實際情況權衡使用。檔案同步功能的開發過程中，常遇到測試困難的問題。透過抽象化設計，將檔案系統操作與核心邏輯分離，可以有效提升測試效率。結合依賴注入，更能簡化測試程式碼，並確保程式碼品質。
Repository 模式與持久化無關模型的探討
在軟體開發中，特別是當我們處理複雜的商業邏輯和資料持久化時，Repository 模式提供了一種將資料儲存和檢索邏輯與業務邏輯分離的有效方法。本文將探討 Repository 模式的實作、優缺點以及其在 Python 中的應用。
Repository 模式的實作
Repository 模式的核心是定義一個介面或抽象類別來封裝資料存取邏輯。這使得我們可以在不修改業務邏輯的情況下，更換不同的資料儲存實作。
使用 SQLAlchemy 實作 Repository
class SqlAlchemyRepository(AbstractRepository):
def __init__(self, session):
self.session = session
def add(self, batch):
self.session.add(batch)
def get(self, reference):
return self.session.query(model.Batch).filter_by(reference=reference).one()
def list(self):
return self.session.query(model.Batch).all()
在上述程式碼中，SqlAlchemyRepository 類別實作了 AbstractRepository 介面，使用 SQLAlchemy 作為 ORM 工具來與資料函式庫互動。
不使用 ORM 的 Repository 實作
練習中提到，可以嘗試不使用 ORM 直接實作 Repository。這需要手動撰寫 SQL 查詢陳述式，並處理資料函式庫連線。
Fake Repository 用於測試
使用 Repository 模式的一個好處是，可以輕易地建立一個 Fake Repository 用於單元測試。
class FakeRepository(AbstractRepository):
def __init__(self, batches):
self._batches = set(batches)
def add(self, batch):
self._batches.add(batch)
def get(self, reference):
return next(b for b in self._batches if b.reference == reference)
def list(self):
return list(self._batches)
內容解密：
__init__方法初始化了一個批次集合，用於儲存batch物件。add方法將新的batch新增到集合中。get方法透過參考值查詢特定的batch。list方法傳回所有批次的列表。
埠和介面卡模式
埠和介面卡模式是一種軟體架構模式，它強調將應用程式的核心邏輯與外部依賴（如資料函式庫）分離。在 Python 中，可以使用抽象基礎類別來定義埠。
內容解密：
AbstractRepository是埠，定義了 Repository 的介面。SqlAlchemyRepository和FakeRepository是介面卡，分別實作了與 SQLAlchemy 和 Fake 資料儲存的互動邏輯。
Repository 模式的權衡
| 優點 | 缺點 |
|---|---|
| 提供了一個簡單的介面用於資料存取 | 需要額外的工作來維護 ORM 對映 |
| 易於建立 Fake Repository 用於測試 | 增加了一層間接性，提高了維護成本 |
| 有助於專注於業務問題 | 對簡單的 CRUD 應用來說可能過於複雜 |
最終，是否採用 Repository 模式取決於應用程式的複雜度和需求。對於簡單的 CRUD 應用，直接使用 ORM 或 ActiveRecord 模式可能就足夠了。但對於複雜的領域模型，Repository 模式提供了一種更好的架構選擇。
儲存函式庫模式回顧與抽象化的重要性
在軟體開發中，保持領域模型與基礎設施關注點的分離至關重要。儲存函式庫模式（Repository Pattern）提供了一種簡單的抽象化方法，來處理永久性儲存的問題。這種模式讓我們能夠將領域模型與特定的資料儲存技術解耦，從而提高程式碼的可測試性和可維護性。
依賴反轉原則的應用
要實作儲存函式庫模式，首先需要應用依賴反轉原則（Dependency Inversion Principle, DIP）。這意味著我們的領域模型應該獨立於基礎設施，如ORM（Object-Relational Mapping）工具。相反，ORM應該依賴於領域模型，而不是相反。這種設計使得領域模型保持乾淨和獨立於外部技術。
儲存函式庫模式的好處
儲存函式庫模式提供了一種在記憶體中操作物件集合的假象，使得對領域模型的測試和操作變得更加直觀和方便。它還使得在不同基礎設銷之間切換變得更加容易，而不會干擾到核心應用程式的邏輯。
抽象化的重要性
抽象化是軟體設計中的一個重要概念。良好的抽象化可以隱藏複雜的細節，提高程式碼的可維護性和可測試性。在本章中，我們將探討什麼是良好的抽象化，以及如何透過抽象化來減少系統之間的耦合度。
耦合度與抽象化
耦合度是指系統中不同模組或元件之間相互依賴的程度。高耦合度可能導致系統難以修改和維護。抽象化可以透過隱藏複雜的細節來減少耦合度，從而提高系統的可維護性和可測試性。
檔案同步範例
讓我們考慮一個檔案同步的例子。我們需要編寫程式碼來同步兩個檔案目錄：來源目錄和目標目錄。需求如下：
- 如果檔案存在於來源目錄但不存在於目標目錄，則將檔案複製到目標目錄。
- 如果檔案存在於來源目錄，但名稱與目標目錄中的名稱不同，則重新命名目標目錄中的檔案以匹配來源目錄中的名稱。
- 如果檔案存在於目標目錄但不存在於來源目錄，則刪除目標目錄中的檔案。
使用雜湊函式進行檔案比較
為了檢測重新命名的檔案，我們需要比較檔案的內容。這可以透過使用雜湊函式（如SHA-1）來實作。下面是一個計算檔案SHA-1雜湊的範例程式碼：
import hashlib
BLOCKSIZE = 65536
def hash_file(path):
hasher = hashlib.sha1()
with path.open("rb") as file:
buf = file.read(BLOCKSIZE)
while buf:
hasher.update(buf)
buf = file.read(BLOCKSIZE)
return hasher.hexdigest()
內容解密：
此程式碼定義了一個函式 hash_file，用於計算指設定檔案的 SHA-1 雜湊值。首先，它初始化了一個 SHA-1 雜湊物件 hasher。然後，它以二進位制讀取模式開啟指定的檔案，並以區塊（大小為 BLOCKSIZE）的形式讀取檔案內容，每讀取一個區塊就更新 hasher。最後，它傳回計算出的雜湊值的十六進位製表示。
基本的同步演算法
我們的初步實作可能如下所示：
def sync(source_dir, target_dir):
# 實作同步邏輯
pass
內容解密：
此函式 sync 是同步演算法的入口點，接受來源目錄和目標目錄作為引數。具體的同步邏輯需要在這個函式中實作，包括比較兩個目錄中的檔案、複製、重新命名或刪除檔案等操作。
透過這種方式，我們可以逐步建立起一個完整的檔案同步系統，並且保持程式碼的可維護性和可測試性。
最佳化檔案同步功能的測試與抽象化設計
在開發檔案同步功能時，我們首先關注的是如何有效地測試這項功能。原始的實作方式將領域邏輯與 I/O 程式碼緊密結合，使得測試變得困難且繁瑣。本文將探討如何透過抽象化設計來簡化測試流程並提升程式碼的可維護性。
原始實作的問題
原始的 sync 函式直接操作檔案系統，導致測試難以進行。測試案例需要建立臨時目錄、寫入檔案，並在測試後清理這些資源。這種方式不僅繁瑣，而且測試速度慢且難以閱讀和維護。
import hashlib
import os
import shutil
from pathlib import Path
def sync(source, dest):
# 原始實作，直接操作檔案系統
source_hashes = {}
for folder, _, files in os.walk(source):
for fn in files:
source_hashes[hash_file(Path(folder) / fn)] = fn
seen = set()
for folder, _, files in os.walk(dest):
for fn in files:
dest_path = Path(folder) / fn
dest_hash = hash_file(dest_path)
seen.add(dest_hash)
if dest_hash not in source_hashes:
dest_path.remove()
elif dest_hash in source_hashes and fn != source_hashes[dest_hash]:
shutil.move(dest_path, Path(folder) / source_hashes[dest_hash])
for src_hash, fn in source_hashes.items():
if src_hash not in seen:
shutil.copy(Path(source) / fn, Path(dest) / fn)
原始測試案例的問題
原始的測試案例需要建立和清理臨時目錄，這使得測試程式碼冗長且難以維護。
def test_when_a_file_exists_in_the_source_but_not_the_destination():
try:
source = tempfile.mkdtemp()
dest = tempfile.mkdtemp()
content = "I am a very useful file"
(Path(source) / 'my-file').write_text(content)
sync(source, dest)
expected_path = Path(dest) / 'my-file'
assert expected_path.exists()
assert expected_path.read_text() == content
finally:
shutil.rmtree(source)
shutil.rmtree(dest)
抽象化設計
為瞭解決上述問題，我們可以將檔案同步的過程抽象化為三個步驟：
- 讀取檔案系統狀態：使用
os.walk遍歷目錄並計算檔案的雜湊值。 - 判斷檔案的狀態（新增、重新命名或冗餘）：比較來源和目標目錄的檔案雜湊值。
- 執行檔案操作（複製、移動或刪除）：根據檔案狀態執行相應的操作。
抽象化後的程式碼結構
我們可以將這些步驟抽象化為不同的函式或類別，以提高程式碼的可測試性和可維護性。
def read_filesystem_state(path):
"""讀取指定路徑的檔案系統狀態，傳回一個字典，鍵為檔案雜湊值，值為檔案路徑。"""
filesystem_state = {}
for folder, _, files in os.walk(path):
for fn in files:
file_path = Path(folder) / fn
filesystem_state[hash_file(file_path)] = str(file_path.relative_to(path))
return filesystem_state
def determine_actions(src_state, dst_state):
"""根據來源和目標的檔案系統狀態，決定需要執行的動作。"""
actions = []
for src_hash, src_path in src_state.items():
if src_hash not in dst_state:
actions.append(("COPY", src_path, src_path))
elif dst_state[src_hash] != src_path:
actions.append(("MOVE", dst_state[src_hash], src_path))
for dst_hash in dst_state.keys() - src_state.keys():
actions.append(("DELETE", dst_state[dst_hash], None))
return actions
def execute_actions(actions, source, dest):
"""執行決定的動作。"""
for action, *paths in actions:
if action == "COPY":
shutil.copy(Path(source) / paths[0], Path(dest) / paths[1])
elif action == "MOVE":
shutil.move(Path(dest) / paths[0], Path(dest) / paths[1])
elif action == "DELETE":
(Path(dest) / paths[0]).unlink()
抽象化後的測試案例
透過抽象化，我們可以簡化測試案例，使其更易於閱讀和維護。
def test_determine_actions_when_a_file_exists_in_the_source_but_not_the_destination():
src_state = {'hash1': 'fn1'}
dst_state = {}
expected_actions = [('COPY', 'fn1', 'fn1')]
assert determine_actions(src_state, dst_state) == expected_actions
測試驅動開發與依賴注入的實踐
在軟體開發中，測試驅動開發（TDD）是一種確保程式碼正確性和可維護性的重要方法。本文將探討如何使用TDD和依賴注入（DI）來改進檔案同步功能的實作。
測試案例：檔案重新命名
def test_when_a_file_has_been_renamed_in_the_source():
src_hashes = {'hash1': 'fn1'}
dst_hashes = {'hash1': 'fn2'}
actions = determine_actions(src_hashes, dst_hashes, Path('/src'), Path('/dst'))
assert list(actions) == [('move', Path('/dst/fn2'), Path('/dst/fn1'))]
內容解密：
src_hashes和dst_hashes分別代表來源和目標資料夾中的檔案雜湊值與檔名對應表。determine_actions函式根據這兩個雜湊表決定需要執行的動作（複製、移動或刪除）。- 在這個測試案例中，來源資料夾中的檔案
fn1對應的雜湊值hash1在目標資料夾中存在，但對應的檔名是fn2，因此預期動作是將fn2重新命名為fn1。
將業務邏輯與I/O分離
為了提高程式碼的可測試性，我們將業務邏輯與I/O操作分離。主要的同步函式 sync 被分解為三個步驟：收集輸入、執行業務邏輯、應用輸出。
def sync(source, dest):
# 步驟1：收集輸入
source_hashes = read_paths_and_hashes(source)
dest_hashes = read_paths_and_hashes(dest)
# 步驟2：執行業務邏輯
actions = determine_actions(source_hashes, dest_hashes, source, dest)
# 步驟3：應用輸出
for action, *paths in actions:
if action == 'copy':
shutil.copyfile(*paths)
if action == 'move':
shutil.move(*paths)
if action == 'delete':
os.remove(paths[0])
內容解密：
read_paths_and_hashes函式負責收集指定資料夾中的檔案雜湊值和檔名。determine_actions函式根據收集到的雜湊值決定需要執行的動作。- 最後，根據
determine_actions傳回的動作列表，執行相應的檔案操作。
使用依賴注入提高可測試性
為了進一步提高可測試性，我們引入依賴注入，將I/O操作抽象化。
def sync(reader, filesystem, source_root, dest_root):
source_hashes = reader(source_root)
dest_hashes = reader(dest_root)
# 省略業務邏輯部分...
for action in actions:
if action == 'copy':
filesystem.copy(destpath, sourcepath)
# 省略其他動作處理...
內容解密：
reader和filesystem是兩個被注入的依賴，分別負責讀取檔案資訊和執行檔案操作。- 這種設計使得我們可以在測試中輕易地替換真實的I/O操作為模擬實作。
使用模擬實作進行測試
class FakeFileSystem(list):
def copy(self, src, dest):
self.append(('COPY', src, dest))
def move(self, src, dest):
self.append(('MOVE', src, dest))
def delete(self, dest):
self.append(('DELETE', dest))
內容解密：
FakeFileSystem是一個模擬的檔案系統，用於在測試中記錄被呼叫的操作。- 這種模擬實作使得我們可以在不實際操作檔案系統的情況下測試
sync函式的行為。