asyncio.timeout: ใส่กรอบเวลาให้งาน async โดยไม่กลืน CancelledError

เวลาเรียก API หรือรอคิวในโค้ด async ถ้าไม่มีกรอบเวลา งานหนึ่งค้างได้ทั้ง request ครับ ตั้งแต่ Python 3.11 มี asyncio.timeout() เป็น async context manager ที่ยกเลิกงานในบล็อกเมื่อหมดเวลา แล้วแปลง CancelledError ภายในเป็น TimeoutError ให้จับนอกบล็อก

WHY: timeout ต้องเป็นขอบเขต

wait_for ยังใช้ได้ แต่ async with asyncio.timeout(seconds): อ่านชัดว่า “ทุกอย่างในนี้มีเพดานเวลา” และรองรับ reschedule() ถ้าเพิ่งรู้ deadline ทีหลัง ตาม เอกสาร asyncio

ข้อสำคัญ: จับ TimeoutError นอก context manager และอย่ากลืน CancelledError ในโค้ดลูก เพราะ structured concurrency พึ่ง cancellation ภายใน

HOW: ทดสอบด้วย IsolatedAsyncioTestCase

import asyncio
import unittest


async def fetch_ok():
    await asyncio.sleep(0.01)
    return "ok"


async def fetch_slow():
    await asyncio.sleep(10)
    return "late"


class TimeoutContextTests(unittest.IsolatedAsyncioTestCase):
    async def test_completes_within_budget(self):
        async with asyncio.timeout(1):
            result = await fetch_ok()
        self.assertEqual(result, "ok")

    async def test_raises_timeout_error(self):
        with self.assertRaises(TimeoutError):
            async with asyncio.timeout(0.05):
                await fetch_slow()


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

จุดที่พลาดบ่อย

• อย่า except CancelledError: pass ในงานลูกแล้วไม่ raise ต่อ
• อย่าจับ TimeoutError ข้างในบล็อก async with asyncio.timeout(...) แล้วคิดว่าจัดการครบ
• งานที่ต้องรอดแม้ caller โดนยกเลิก ให้คิด ownership ใหม่ ไม่ใช่แค่ซ่อน exception

Recap

• ใช้ asyncio.timeout เมื่อต้องการเพดานเวลาเป็นบล็อก
• จับ TimeoutError นอก context manager
• ทดสอบทั้ง happy path และ timeout ด้วย IsolatedAsyncioTestCase
• อ่านเอกสาร timeouts ของ asyncio ก่อนย้ายจาก wait_for ครับ