Python contextlib.aclosing: ปิด Async Generator แม้ Loop หยุดก่อนจบ

async generator มักถือ cursor, connection หรือ subscription ไว้ ถ้า consumer break หลังเจอข้อมูลที่ต้องการ finally ภายใน generator ต้องยังทำงาน contextlib.aclosing ครอบ object ที่มี aclose() และเรียกปิดตอนออกจาก async with แม้ loop ไม่ได้อ่านจนหมดครับ

จุดสำคัญคือ cleanup รันใน async context เดียวกับ iteration จึงยังเห็น context variables และ exception handling เดิม ไม่ถูกเลื่อนไปทำหลัง task จบ

from __future__ import annotations

import unittest
from collections.abc import AsyncIterator
from contextlib import aclosing


async def events(closed: list[bool]) -> AsyncIterator[int]:
    try:
        for value in range(3):
            yield value
    finally:
        closed.append(True)


async def first_event(closed: list[bool]) -> int:
    async with aclosing(events(closed)) as stream:
        async for value in stream:
            return value
    raise LookupError("empty stream")


class AclosingTests(unittest.IsolatedAsyncioTestCase):
    async def test_closes_after_early_return(self):
        closed: list[bool] = []
        self.assertEqual(await first_event(closed), 0)
        self.assertEqual(closed, [True])


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

ใช้ aclosing เมื่อ object มี aclose() แต่ไม่ได้เป็น async context manager อยู่แล้ว ถ้า API คืน context manager โดยตรง ให้ใช้ protocol ของมันแทนเพื่อไม่ซ้อน lifecycle เกินจำเป็น และอย่ากลืน exception ใน finally เพราะจะทำให้ต้นเหตุจาก loop หายไป

Recap

ครอบ async generator ที่อาจถูกหยุดก่อนหมดด้วย aclosing
วาง cleanup ใน finally ของ generator
เทสต์ break, early return และ exception path
ใช้ context manager เดิมของ resource หากมีอยู่แล้ว

Recap

เลือกการใช้คุกกี้วิเคราะห์