01Object model และ Mutability

ตัวแปรใน Python คือชื่อที่อ้างถึง object ไม่ใช่กล่องที่เก็บค่าโดยตรง การ assign จึงเปลี่ยน reference ส่วนการ mutate เปลี่ยน object เดิม ความแตกต่างนี้อธิบาย bug ที่พบบ่อย เช่น default argument แบบ mutable ซึ่งถูกสร้างครั้งเดียวตอนประกาศฟังก์ชัน

Immutable

`int`, `str`, `tuple` และ `frozenset` เปลี่ยนค่าภายในไม่ได้ operation มักคืน object ใหม่

Mutable

`list`, `dict`, `set` และ instance ทั่วไปเปลี่ยน state เดิมได้ ต้องระวัง shared reference

02Typing และ Error boundaries

Type hints ช่วย static analysis, IDE และสื่อสาร contract แต่ไม่ validate ข้อมูลตอน runtime โดยอัตโนมัติ ข้อมูลที่เข้าจาก HTTP, queue หรือ database จึงยังต้อง parse และ validate ที่ boundary ก่อนเปลี่ยนเป็น domain object

ใช้ exception กับเหตุการณ์ผิดปกติ ไม่ใช้ควบคุม flow ปกติ จับ exception เฉพาะชนิดที่จัดการได้ เพิ่มบริบทด้วย `raise ... from err` และ log หนึ่งครั้งตรง application boundary เพื่อหลีกเลี่ยง stack trace ซ้ำหลายชั้น

  • ใช้ `dataclass` หรือ immutable model เพื่อสื่อความหมายของข้อมูล
  • ใช้ Protocol เมื่อต้องการ structural typing และลด coupling
  • อย่าปล่อย `Any` ไหลผ่าน core domain เพราะจะเสียประโยชน์ของ type checker

03asyncio, Thread และ Process

`asyncio` เหมาะกับ I/O concurrency เมื่อ library รองรับ async ส่วน thread เหมาะกับ blocking I/O ที่เปลี่ยนไม่ได้ และ process เหมาะกับ CPU-bound work ที่ต้องกระจายข้าม interpreter process อย่าใส่ `async` ให้ฟังก์ชัน CPU-heavy แล้วคาดว่าจะเร็วขึ้น

policy_batch.py
import asyncio

async def fetch_policy(policy_id: str, semaphore: asyncio.Semaphore):
    async with semaphore:  # จำกัดงานที่ทำพร้อมกัน
        async with asyncio.timeout(1.0):
            return await policy_client.get(policy_id)

async def fetch_all(policy_ids: list[str]):
    semaphore = asyncio.Semaphore(20)

    async with asyncio.TaskGroup() as group:
        tasks = [
            group.create_task(fetch_policy(pid, semaphore))
            for pid in policy_ids
        ]

    return [task.result() for task in tasks]

`TaskGroup` ให้ structured concurrency: เมื่อ task หนึ่งล้มเหลว task อื่นในกลุ่มจะถูกยกเลิกและ exception ถูกรวบรวมอย่างเป็นระบบ แต่ยังต้องจำกัด concurrency ด้วย semaphore ไม่เช่นนั้นการสร้าง task หลายหมื่นตัวอาจถล่ม downstream ได้

04Performance และ Production readiness

เริ่มจาก profiling ก่อน optimize แยกให้ออกว่าเวลาหายไปกับ CPU, I/O, allocation หรือ serialization สำหรับ API ให้ดู p95/p99 latency, event-loop lag, connection pool และจำนวน task ที่ค้าง ไม่ใช่ดู average อย่างเดียว

อ่านเอกสารทางการ: Python asyncio Tasks และ TaskGroup