Python 介面設計：以 Function 簡化 Class Hook

– 8 分鐘閱讀

Python 提供多種客製化行為的方式，其中以 Function 作為介面最為簡潔優雅。許多內建 API 允許傳入 Function 調整行為，這些 Function 如同鉤子 (Hook)，在 API 執行時被呼叫，讓開發者得以介入流程。例如，list 的 sort 方法可傳入 key 引數，決定排序依據。Python 的 Function 是一級公民，能像其他值一樣傳遞和參照，因此能勝任 Hook 角色。相較於 Class，Function 更易於理解和說明，定義也更簡單。defaultdict 即是一個允許提供 Function 作為 Hook 的例子，可在存取不存在的 Key 時被呼叫並回傳預設值。

為追蹤 defaultdict 遺失 Key 的次數，可使用 Stateful Closure。然而，Closure 的可讀性較差，更好的方法是用 Class 封裝狀態。雖然其他語言可能需要修改 defaultdict 才能相容 Class 介面，但在 Python 中，可以直接參照 Class 的方法傳給 defaultdict 作為 Hook。為提升 Class 的意圖明確性，Python 的 __call__ 特殊方法讓物件像 Function 一樣被呼叫，與 callable 內建函式會對此類別物件回傳 True。__call__ 方法表明 Class 例項可作為 Function 引數使用，例如 API Hook。它引導讀者找到 Class 主要行為的入口點，也暗示 Class 的目標是作為 Stateful Closure。

defaultdict 只要求一個 Function 作為預設值 Hook，Python 提供多種方式滿足此需求，例如 lambda 表示式、Stateful Closure、Class 方法或帶有 __call__ 方法的 Class 例項，開發者可根據需求選擇。

玄貓解說：Python 介面設計的極簡之道 - 告別 Class，擁抱 Function

在 Python 的世界裡，客製化行為的方式有很多種，但最簡潔優雅的，莫過於使用 Function 作為介面。許多內建 API 都允許你傳入 Function 來調整行為，這些 Function 就像鉤子 (Hook)，在 API 執行時被呼叫，讓你的程式碼得以介入。

例如，list 的 sort 方法，可以傳入 key 引數，決定排序的依據。以下範例，玄貓用 lambda 表示式，依據字串長度排序名字：

names = ['Socrates', 'Archimedes', 'Plato', 'Aristotle']
names.sort(key=lambda x: len(x))
print(names)
# 輸出：['Plato', 'Socrates', 'Aristotle', 'Archimedes']

捨棄 Class 的理由：Function 的優勢

在其他語言，你可能會用抽象 Class 定義 Hook。但在 Python，許多 Hook 只是無狀態的 Function，定義好引數與回傳值即可。Function 作為 Hook 的優勢在於：

  • 易於描述： Function 比 Class 更容易理解和說明。
  • 簡潔： Function 的定義比 Class 更簡單。

Function 能勝任 Hook 的原因，在於 Python 的 Function 具有「一級公民」 (First-Class Function) 的身分：Function 和 Method 可以像其他 Value 一樣傳遞和參照。

defaultdict 的妙用：Function Hook 示範

舉例來說，defaultdict 允許你提供一個 Function，在存取不存在的 Key 時被呼叫，並回傳預設值。

以下玄貓定義一個 Hook，記錄每次 Key 遺失的事件，並回傳 0 作為預設值：

def log_missing():
    print('Key added')
    return 0

給定一個初始 Dictionary 和一組欲增加的專案，log_missing Function 會執行兩次 (針對 ‘red’ 和 ‘orange’)：

from collections import defaultdict

current = {'green': 12, 'blue': 3}
increments = [
    ('red', 5),
    ('blue', 17),
    ('orange', 9),
]

result = defaultdict(log_missing, current)
print('Before:', dict(result))
for key, amount in increments:
    result[key] += amount
print('After: ', dict(result))

內容解密

  1. from collections import defaultdict: 匯入 defaultdict，它是一個在存取不存在的鍵時提供預設值的字典。
  2. current = {'green': 12, 'blue': 3}: 定義一個包含現有鍵和值的字典。
  3. increments = [('red', 5), ('blue', 17), ('orange', 9)]: 定義一個包含要增加到 current 字典中的鍵和值的列表。
  4. result = defaultdict(log_missing, current): 建立一個 defaultdict 例項，其中 log_missing 函式用於提供預設值。
  5. print('Before:', dict(result)): 列印預 result 字典的初始狀態。
  6. for key, amount in increments:: 遍歷 increments 列表。
  7. result[key] += amount: 將 amount 增加到 result 字典中 key 鍵對應的值。如果 key 不存在，log_missing 函式將被呼叫，傳回預設值 0，然後再加上 amount。
  8. print('After: ', dict(result)): 列印預 result 字典的最終狀態。

執行結果

Before: {'green': 12, 'blue': 3}
Key added
Key added
After:  {'green': 12, 'blue': 20, 'red': 5, 'orange': 9}

Stateful Closure：追蹤狀態的 Hook

如果想追蹤 defaultdict 遺失 Key 的總數，可以用 Stateful Closure：

def increment_with_report(current, increments):
    added_count = 0
    def missing():
        nonlocal added_count  # Stateful closure
        added_count += 1
        return 0
    result = defaultdict(missing, current)
    for key, amount in increments:
        result[key] += amount
    return result, added_count

內容解密

  1. def increment_with_report(current, increments):: 定義一個函式，該函式接受當前字典和增量列表作為引數。
  2. added_count = 0: 初始化一個變數來追蹤新增的鍵的數量。
  3. def missing():: 定義一個內部函式，該函式將在存取缺失鍵時被呼叫。
  4. nonlocal added_count: 允許 missing 函式修改外部函式 increment_with_report 中定義的 added_count 變數。
  5. added_count += 1: 增加 added_count 變數的值。
  6. return 0: 傳回 0 作為缺失鍵的預設值。
  7. result = defaultdict(missing, current): 建立一個 defaultdict 例項，其中 missing 函式用於提供預設值。
  8. for key, amount in increments:: 遍歷 increments 列表。
  9. result[key] += amount: 將 amount 增加到 result 字典中 key 鍵對應的值。如果 key 不存在，missing 函式將被呼叫，傳回預設值 0，然後再加上 amount。
  10. return result, added_count: 傳回更新後的字典和新增的鍵的數量。

即使 defaultdict 不知道 missing Hook 維護了狀態，也能得到預期的結果 (2)。

result, count = increment_with_report(current, increments)
assert count == 2

Class 的逆襲：封裝狀態

用 Closure 實作 Stateful Hook 的問題是，程式碼可讀性較差。另一種方法是用 Class 封裝要追蹤的狀態：

class CountMissing(object):
    def __init__(self):
        self.added = 0
    def missing(self):
        self.added += 1
        return 0

在其他語言，你可能需要修改 defaultdict 才能相容 CountMissing 的介面。但在 Python，由於 Function 是一級公民，你可以直接參照 CountMissing.missing 方法，傳給 defaultdict 作為預設值 Hook。

counter = CountMissing()
result = defaultdict(counter.missing, current)  # Method ref
for key, amount in increments:
    result[key] += amount
assert counter.added == 2

__call__ 的魔法：讓物件像 Function 一樣被呼叫

為了讓 Class 的意圖更明確，Python 允許 Class 定義 __call__ 特殊方法。__call__ 讓物件可以像 Function 一樣被呼叫，與 callable 內建 Function 會對此類別物件回傳 True。

class BetterCountMissing(object):
    def __init__(self):
        self.added = 0
    def __call__(self):
        self.added += 1
        return 0

counter = BetterCountMissing()
counter()
assert callable(counter)

現在，玄貓用 BetterCountMissing 例項作為 defaultdict 的預設值 Hook，追蹤新增 Key 的數量：

counter = BetterCountMissing()
result = defaultdict(counter, current)  # Relies on __call__
for key, amount in increments:
    result[key] += amount
assert counter.added == 2

__call__ 方法表明，Class 的例項可以用在需要 Function 引數的地方 (例如 API Hook)。它引導讀者找到負責 Class 主要行為的入口點，並暗示 Class 的目標是作為 Stateful Closure。

總之，defaultdict 並不關心你用了什麼技巧，它只要求一個 Function 作為預設值 Hook。Python 提供了多種方式來滿足這個簡單的 Function 介面，端看你的需求而定。

玄貓的重點整理

  • 對於簡單的介面，Function 通常比 Class 更適合。
  • Python 的 Function 和 Method 都是一級公民，可以像 Value 一樣傳遞。
  • 善用 __call__ 方法，讓 Class 的例項可以像 Function 一樣被呼叫，增加程式碼的可讀性。
  • 在設計 API 時，優先考慮使用 Function 作為 Hook，讓使用者可以更靈活地客製化行為。

在 Python 中，__call__ 特殊方法賦予類別的例項如同普通函式般被呼叫的能力。

當你需要一個函式來維持狀態時，考慮定義一個提供 __call__ 方法的類別，而不是定義一個有狀態的閉包（參見條目 15：「瞭解閉包如何與變數作用域互動」）。

活用 @classmethod 多型來通用地建構物件

在 Python 中，不僅物件支援多型，類別也支援。這代表什麼？又有什麼好處？

多型是一種層級結構中的多個類別實作其自身獨特版本方法的方式。這使得許多類別能夠實作相同的介面或抽象基底類別，同時提供不同的功能（範例請參見條目 28：「從 collections.abc 繼承以取得自訂容器型別」）。

舉例來說，假設你正在編寫一個 MapReduce 實作，並且你需要一個通用的類別來表示輸入資料。我在此定義這樣一個類別，其 read 方法必須由子類別定義：

class InputData(object):
    def read(self):
        raise NotImplementedError

接著，我建立一個 InputData 的具體子類別，它從磁碟上的檔案讀取資料：

class PathInputData(InputData):
    def __init__(self, path):
        super().__init__()
        self.path = path

    def read(self):
        return open(self.path).read()

你可以擁有任意數量的 InputData 子類別，例如 PathInputData，它們都可以實作 read 的標準介面，以傳回要處理的資料位元組。其他 InputData 子類別可以從網路讀取、透明地解壓縮資料等等。

你會希望 MapReduce worker 也有類別似的抽象介面，以標準方式使用輸入資料。

class Worker(object):
    def __init__(self, input_data):
        self.input_data = input_data
        self.result = None

    def map(self):
        raise NotImplementedError

    def reduce(self, other):
        raise NotImplementedError

接著，我定義一個 Worker 的具體子類別，以實作我想套用的特定 MapReduce 函式：一個簡單的換行符計數器：

class LineCountWorker(Worker):
    def map(self):
        data = self.input_data.read()
        self.result = data.count('\n')

    def reduce(self, other):
        self.result += other.result

這個實作看起來進展順利，但我遇到了所有這些步驟中最大的障礙。是什麼將所有這些部分連線起來？我有一組不錯的類別，具有合理的介面和抽象——但只有在建構物件之後才有用。誰負責建構物件並協調 MapReduce？

最簡單的方法是手動建構物件並使用一些輔助函式連線它們。我在此列出目錄的內容，並為其包含的每個檔案建構一個 PathInputData 例項：

import os

def generate_inputs(data_dir):
    for name in os.listdir(data_dir):
        yield PathInputData(os.path.join(data_dir, name))

接下來，我使用 generate_inputs 傳回的 InputData 例項建立 LineCountWorker 例項。

def create_workers(input_list):
    workers = []
    for input_data in input_list:
        workers.append(LineCountWorker(input_data))
    return workers

我透過將 map 步驟分散到多個執行緒來執行這些 Worker 例項（參見條目 37：「使用執行緒進行封鎖 I/O，避免用於平行處理」）。然後，我重複呼叫 reduce 以將結果合併為一個最終值。

from threading import Thread

def execute(workers):
    threads = [Thread(target=w.map) for w in workers]
    for thread in threads: thread.start()
    for thread in threads: thread.join()

    first, rest = workers[0], workers[1:]
    for worker in rest:
        first.reduce(worker)
    return first.result

最後，我將所有部分連線在一個函式中以執行每個步驟。

def mapreduce(data_dir):
    inputs = generate_inputs(data_dir)
    workers = create_workers(inputs)
    return execute(workers)

在測試輸入檔案集上執行此函式效果很好。

from tempfile import TemporaryDirectory

def write_test_files(tmpdir):
    # 這裡省略了建立測試檔案的程式碼，因為它與主題無關
    pass

with TemporaryDirectory() as tmpdir:
    write_test_files(tmpdir)
    result = mapreduce(tmpdir)
    print('There are', result, 'lines')

>>>
There are 4360 lines

問題是什麼？最大的問題是 mapreduce 函式根本不通用。如果你想編寫另一個 InputData 或 Worker 子類別，你也必須重寫 generate_inputs、create_workers 和 mapreduce 函式才能比對。

這個問題歸結為需要一種通用的物件建構方式。在其他語言中，你可以使用建構函式多型來解決這個問題，要求每個 InputData 子類別提供一個特殊的建構函式，輔助方法可以使用該建構函式來協調 MapReduce。問題是 Python 只允許使用單一建構函式方法 __init__。要求每個 InputData 子類別都具有相容的建構函式是不合理的。

解決這個問題的最佳方法是使用 @classmethod 多型。這與我用於 InputData.read 的例項方法多型完全相同，只是它適用於整個類別，而不是它們建構的物件。

讓我將這個想法應用於 MapReduce 類別。我在此使用一個泛型類別方法擴充 InputData 類別，該方法負責使用通用介面建立新的 InputData 例項：

class GenericInputData(object):
    def read(self):
        raise NotImplementedError

    @classmethod
    def generate_inputs(cls, config):
        raise NotImplementedError

玄貓認為 generate_inputs 採用一個包含一組組態引數的字典，這些引數由 InputData 具體子類別來解譯。我在此使用組態來尋找

玄貓

玄貓

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