Python AsyncExitStack: รวม cleanup แบบ sync และ async ไว้ในจุดเดียว

resource ของงาน async ไม่ได้เปิดครบทุกครั้ง บางรอบมี connection บางรอบมี lock เพิ่ม และบางชิ้นปิดด้วยฟังก์ชันธรรมดาครับ AsyncExitStack เก็บ cleanup ที่เกิดขึ้นจริง แล้วเรียกย้อนลำดับเมื่อออกจาก scope

WHY: เปิดสำเร็จเท่าไร ปิดเท่านั้น

เหมือนฝากกุญแจทุกดอกไว้ที่เคาน์เตอร์ตามลำดับที่รับ พอเลิกงานพนักงานคืนจากดอกล่าสุดย้อนกลับไปดอกแรก ถ้าเปิด resource ชิ้นที่สามไม่สำเร็จ สองชิ้นแรกก็ยังถูกปิด ไม่ต้องเขียน try/finally ซ้อนหลายชั้น

import unittest
from contextlib import AsyncExitStack, asynccontextmanager


@asynccontextmanager
async def resource(name, events):
    events.append(f"open:{name}")
    try:
        yield name
    finally:
        events.append(f"close:{name}")


class AsyncExitStackTests(unittest.IsolatedAsyncioTestCase):
    async def test_closes_in_reverse_order(self):
        events = []
        async with AsyncExitStack() as stack:
            await stack.enter_async_context(resource("db", events))
            await stack.enter_async_context(resource("lock", events))
            self.assertEqual(events, ["open:db", "open:lock"])

        self.assertEqual(
            events,
            ["open:db", "open:lock", "close:lock", "close:db"],
        )


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

ใช้ enter_async_context กับ async context manager, push_async_callback กับ coroutine cleanup และ callback กับงาน sync อย่าส่ง callback ที่อาจ block event loop นาน ๆ; ย้าย I/O แบบ sync ไป thread หรือใช้ async API ที่แท้จริง

Recap

• ลงทะเบียน cleanup ทันทีหลังเปิด resource สำเร็จ
• stack ปิดย้อนลำดับเหมือน with ที่ซ้อนกัน
• ใช้ได้กับ resource ที่เปิดตามเงื่อนไข
• อย่า block event loop ด้วย callback แบบ sync ที่ช้า