งูต้นฉบับกำลังคลุมกรอบงาน async ด้วยผ้าสีครีมที่ใช้ได้ทั้งรอบกล่องและรอบสายพาน

Python AsyncContextDecorator: ใช้ lifecycle เดียวกันทั้ง async with และ decorator

พี่โชว์Engineering

เมื่อ lifecycle เดียวกันต้องครอบหลาย async function การเขียน async with ซ้ำทุกจุดทำให้โค้ดยาวและลืม cleanup ได้ AsyncContextDecorator ให้ context manager เดียวใช้เป็น decorator ได้ด้วยครับ

from __future__ import annotations

import unittest
from contextlib import AsyncContextDecorator


class audit_scope(AsyncContextDecorator):
    def __init__(self, events: list[str]) -> None:
        self.events = events

    async def __aenter__(self) -> "audit_scope":
        self.events.append("enter")
        return self

    async def __aexit__(self, exc_type, exc, tb) -> bool:
        self.events.append("exit")
        return False


class AuditScopeTests(unittest.IsolatedAsyncioTestCase):
    async def test_decorator_preserves_lifecycle(self) -> None:
        events: list[str] = []

        @audit_scope(events)
        async def work() -> str:
            events.append("work")
            return "ok"

        self.assertEqual(await work(), "ok")
        self.assertEqual(events, ["enter", "work", "exit"])


if __name__ == "__main__":
    unittest.main()

เหมือนบัตรผ่านชั่วคราวที่แตะเข้าและแตะออกเองเมื่อพนักงานเริ่มงาน Decorator ช่วยลด ceremony แต่ไม่เปลี่ยนกฎ exception: คืน False จาก __aexit__ เพื่อปล่อย error เดิมออกไป คืน True เฉพาะเมื่อออกแบบให้กลืน exception นั้นจริง ๆ

Context ต้องใช้ซ้ำได้เมื่อ decorator ถูกเรียกหลายครั้ง ถ้ามี state ต่อ invocation ให้ override _recreate_cm() เพื่อสร้าง instance ใหม่ หรือเก็บ state ในตัวแปร context-local ไม่เช่นนั้นงานพร้อมกันอาจเหยียบ state กัน

Recap

  • ใช้ lifecycle เดียวกันได้ทั้ง async with และ decorator
  • ปล่อย exception เดิมด้วย False เป็นค่าเริ่มต้น
  • ตรวจ reentrancy และ concurrent calls
  • ทดสอบทั้ง success และ exception path