デコレーターとコンテキストマネージャー

デコレーターはcallableに上着を着せる

関数は()で呼べるcallableです。デコレーターは、callableを受け取り、利用者が呼ぶcallableを返す普通のPythonです。

def announce(function):
    def wrapper(*args, **kwargs):
        print("前")
        result = function(*args, **kwargs)
        print("後")
        return result

    return wrapper


@announce
def greeting(name):
    return f"こんにちは、{name}"


print(greeting("ミナ"))

出力：

前
後
こんにちは、ミナ

@announceはgreeting = announce(greeting)という意味です。Pythonが関数定義を実行するときに装飾し、後で呼び出すたびにwrapperが動きます。透過的なwrapperは*argsと**kwargsを転送し、元の結果を返します。return resultを忘れると、呼び出し側は静かにNoneを受け取ります。

名前を守り、上着を設定する

普通のwrapperの名前は、greetingではなくwrapperです。functools.wraps()は元の名前、ドキュメント、アノテーション、__wrapped__へのリンクを守ります。

設定できるデコレーターは、三つの小さな層を持つデコレーターファクトリーです。

from functools import wraps


def tagged(label):                       # 設定を受け取る
    if not label:
        raise ValueError("ラベルを空にはできません")

    def decorate(function):              # callableを受け取る
        @wraps(function)
        def wrapper(*args, **kwargs):     # 呼び出し引数を受け取る
            return f"[{label}] {function(*args, **kwargs)}"

        return wrapper

    return decorate


@tagged("学習")
def total(a, b):
    """二つの学習時間を足す。"""
    return a + b


print(total(20, 15))
print(total.__name__)

出力：

[学習] 35
total

tagged("学習")とdecorate(total)は定義時に動きます。内側のwrapperは呼び出し時に動きます。デコレーターを重ねると、関数に近い物から適用されます。順番を簡単に説明できる場合だけ重ねましょう。

withは入口と出口を作る

コンテキストマネージャーは、一回の呼び出しではなくブロックを囲みます。withは__enter__()を呼び、ブロックを動かし、その後は必ず__exit__()を呼びます。

class Notebook:
    def __enter__(self):
        print("開く")
        self.notes = []
        return self.notes

    def __exit__(self, exc_type, exc_value, traceback):
        print("閉じる")
        return False


with Notebook() as notes:
    notes.append("コンテキストマネージャー")
    print(notes)

出力：

開く
['コンテキストマネージャー']
閉じる

__enter__()の戻り値がnotesになります。成功時、__exit__()の三つの例外引数はNoneです。失敗時は例外を説明します。truthyな戻り値は例外を抑制し、FalseまたはNoneなら外へ伝えます。抑制は、偶然のTrueではなく、狭く明記した選択にします。

contextlibで入口を小さく書く

入口と出口が一つずつなら、contextlib.contextmanagerがジェネレーターをコンテキストマネージャーに変えます。

from contextlib import contextmanager


@contextmanager
def section(name):
    print("開く", name)
    try:
        yield []
    finally:
        print("閉じる", name)


with section("テスト") as notes:
    notes.append("準備・実行・検証")
    print(notes)

出力：

開く テスト
['準備・実行・検証']
閉じる テスト

yieldより前が入口、yieldされた値がブロックの物、finallyが必ず行う後片付けです。ブロックが失敗すると、例外はyieldの場所へ届き、通常は後片付け後も外へ進みます。広く捕捉するのは、明確な監査やロールバック方針があるときだけです。その後は引数なしのraiseで元のトレースバックを守ります。

短い境界：マネージャーが所有するリソースだけを閉じ、リソース自身のwith対応を優先します。ジェネレーター型マネージャーの一つのインスタンスは一回限りなので、withごとに新しく作ります。

トランザクション型の読書記録を作る

次の標準ライブラリだけの完全なプロジェクトをreading_journal.pyとして保存します。

from contextlib import contextmanager
from dataclasses import dataclass, field
from functools import wraps


@dataclass
class Journal:
    entries: list[str] = field(default_factory=list)
    audit: list[str] = field(default_factory=list)


def audited(audit, label):
    def decorate(function):
        @wraps(function)
        def wrapper(*args, **kwargs):
            audit.append(f"{label}:start")
            try:
                result = function(*args, **kwargs)
            except Exception as error:
                audit.append(f"{label}:error:{type(error).__name__}")
                raise
            else:
                audit.append(f"{label}:ok")
                return result

        return wrapper

    return decorate


@contextmanager
def staged(journal):
    pending = []
    journal.audit.append("batch:open")
    try:
        yield pending
    except Exception:
        journal.audit.append("batch:rollback")
        raise
    else:
        journal.entries.extend(pending)
        journal.audit.append(f"batch:commit:{len(pending)}")
    finally:
        journal.audit.append("batch:close")


def parse_entry(line):
    topic, separator, minutes_text = line.partition("|")
    if not separator or not topic.strip():
        raise ValueError("topic|minutesの形が必要です")
    try:
        minutes = int(minutes_text)
    except ValueError as cause:
        raise ValueError("分数は整数で指定してください") from cause
    if minutes <= 0:
        raise ValueError("分数は正である必要があります")
    return f"{topic.strip()}（{minutes}分）"


def make_importer(journal):
    @audited(journal.audit, "import")
    def import_lines(lines):
        with staged(journal) as pending:
            for line in lines:
                pending.append(parse_entry(line))
        return len(pending)

    return import_lines


journal = Journal()
import_lines = make_importer(journal)

count = import_lines(["デコレーター | 25", "コンテキスト | 30"])
saved = journal.entries.copy()
print("取込：", count)

try:
    import_lines(["正常 | 10", "不正 | many"])
except ValueError as error:
    print("拒否：", error)

assert journal.entries == saved
assert import_lines.__name__ == "import_lines"
print("記録：", journal.entries)
print("最後の監査：", journal.audit[-5:])

python3 reading_journal.pyで実行します。

取込： 2
拒否： 分数は整数で指定してください
記録： ['デコレーター（25分）', 'コンテキスト（30分）']
最後の監査： ['import:start', 'batch:open', 'batch:rollback', 'batch:close', 'import:error:ValueError']

デコレーターは呼び出し単位の監査を所有します。コンテキストマネージャーは仮データ、確定、ロールバック、後片付けを所有します。広い捕捉には一つの明記した仕事があり、すぐ再送出します。失敗したバッチは既存記録を変えません。

3つの小さなチャレンジ

1. docstringを持つ関数を装飾し、結果、__name__、__doc__を確認する。
2. @repeat(times)で0を装飾時に拒否し、関数をちょうど三回呼ぶ。
3. Notebook.__exit__()をKeyErrorだけ抑制する形へ変え、ValueErrorは外へ出ることを証明する。

第19章へ進めるかな？

• デコレーターが普通のcallable変換だと分かる
• wrapperで引数を転送し、結果を返し、@wrapsを使える
• デコレーターファクトリーの三層を説明できる
• __enter__()が返す物と、__exit__()が動く時を知っている
• 必ず行う後片付けをfinallyへ置き、例外抑制を意図して選べる
• 読書記録を実行し、失敗したバッチのロールバックを証明した

次章では、テスト、デバッガー、注意して設定したログという三種類の証拠を集めます。