Context Managers

ปัญหาที่ context manager แก้

หลาย operation ต้องมาคู่กับขั้นตอน cleanup: เปิดไฟล์แล้วปิดมัน, acquire lock แล้ว release มัน, เริ่มจับเวลาแล้วหยุดมัน ถ้ามี exception เกิดขึ้นระหว่างสองขั้นตอนนี้ cleanup จะไม่ทำงานเลย — นำไปสู่ resource leak

statement with แก้ปัญหานี้โดยรับประกันว่า cleanup จะทำงานไม่ว่าจะเกิดอะไรขึ้นก็ตาม

with open("data.txt") as f:
    content = f.read()
# f is always closed here, even if read() raised

`with` ทำงานอย่างไร: protocol

object ใดก็ตามที่มี method __enter__ และ __exit__ คือ context manager

with expr as var:
    body

เทียบเท่ากับ:

import sys

cm = expr
var = cm.__enter__()
try:
    body
except:
    if not cm.__exit__(*sys.exc_info()):
        raise
else:
    cm.__exit__(None, None, None)

__enter__ เตรียม resource แล้วคืนค่าที่ผูกกับ var __exit__ ทำงานเสมอตอนออก ถ้ามันคืนค่า truthy จะ suppress exception; คืน False (หรือ None) จะปล่อยให้ exception propagate ต่อไป bare except: ด้านบนเป็นหนึ่งในไม่กี่กรณีที่การจับทุกอย่างถูกต้อง: if not ...: raise จะ re-raise เว้นแต่ __exit__ จะ suppress อย่างชัดเจน ดังนั้นจึงไม่มี exception ไหนถูกกลืนแบบเงียบ ๆ เลย

flowchart TD
A["cm.__enter__()"] --> B["run the with body"]
B -->|"no exception"| C["cm.__exit__(None, None, None)"]
B -->|"exception raised"| D["cm.__exit__(exc_type, exc, tb)"]
D -->|"returns truthy"| E["exception suppressed -- execution continues after with"]
D -->|"returns falsy"| F["exception re-raised -- propagates past the with"]

Protocol ของ with statement: return value ของ __exit__ ตัดสินว่า exception จะรอดหรือไม่

เขียน class-based context manager

from types import TracebackType
from typing import Self

class Timer:
    def __enter__(self) -> Self:
        print("Timer started")
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> bool:
        print("Timer stopped")
        suppress = False   # do not suppress exceptions
        return suppress

argument ทั้งสามของ __exit__ คือ exception type, value และ traceback — เป็น None ทั้งหมดเมื่อไม่มี exception เกิดขึ้น

`exit` คืน `True` กลืน exception

บรรทัดที่ส่งผลมากที่สุดใน __exit__ implementation ใด ๆ คือ return value ของมัน การคืนค่า truthy ใด ๆ — รวมถึง return True ที่พิมพ์ผิดแทน return False — บอก with statement ว่า exception ถูก handle แล้ว และมันก็หายไป เฉย ๆ:

class Swallower:
    def __enter__(self) -> Self:
        return self
    def __exit__(self, exc_type, exc, tb) -> bool:
        if exc is not None:
            print(f"swallowing {type(exc).__name__}: {exc}")
        return True   # BUG-PRONE if unintentional: any exception disappears here

with Swallower():
    raise ValueError("this never propagates")
print("execution continues here -- the ValueError above is gone")

บางครั้งนี่คือสิ่งที่ต้องการพอดี (contextlib.suppress ด้านล่างสร้างจากพฤติกรรมนี้ทั้งหมด) — แต่ __exit__ ที่คืน True แบบไม่มีเงื่อนไข ซึ่งเขียนขึ้นโดยบังเอิญ ไม่ใช่ตั้งใจ จะลบทุก exception ที่ raise ภายในทุก with block ที่ใช้ มันแบบเงียบ ๆ คืน False (หรือ None หรือไม่คืนอะไรเลย) เสมอ เว้นแต่ตั้งใจจะ suppress จริง ๆ

`contextlib.contextmanager` — ทางลัดด้วย generator

การเขียน class เต็มรูปแบบแค่เพื่อห่อ setup/teardown นั้นยืดยาว contextlib.contextmanager ให้ใช้ generator function แทนได้:

import contextlib

@contextlib.contextmanager
def managed_resource(name: str):
    print(f"Acquiring {name}")
    try:
        yield name          # value bound to the as-variable
    finally:
        print(f"Releasing {name}")

ทุกอย่างก่อน yield คือ logic ของ __enter__ ทุกอย่างหลัง yield (ใน finally) คือ logic ของ __exit__ try/finally ภายใน generator รับประกันว่า teardown จะทำงานแม้มี exception ถูก raise ภายใน with block

`contextlib.suppress` — `exit` ที่คืน `True` แบบมีระเบียบ

contextlib.suppress(*exception_types) คือ context manager ที่ __exit__ คืน True — แต่เฉพาะเมื่อ exception ที่ raise ตรงกับ type ที่ระบุไว้เท่านั้น:

import contextlib

with contextlib.suppress(FileNotFoundError):
    open("does-not-exist.txt")
# execution continues here either way

มันคือเวอร์ชันมีระเบียบของตัวอย่าง Swallower ข้างบน: แทนที่จะเป็น __exit__ ที่กลืน ทุกอย่าง มันกลืนเฉพาะ exception type ที่ระบุไว้เท่านั้น ส่วนที่เหลือ propagate เหมือนเดิมทุกประการ ดูบท Exceptions สำหรับการเทียบกับ try/except/pass

`ExitStack` — จำนวน context manager ที่ไม่แน่นอน

การซ้อน with statement ใช้ได้เมื่อรู้จำนวน resource ล่วงหน้า contextlib.ExitStack จัดการ context manager จำนวนหนึ่งที่ตัดสินใจตอน runtime — เช่นเปิดไฟล์ตาม list ที่ยาวไม่แน่นอน — และรับประกันว่าทั้งหมด unwind ในลำดับ ย้อนกลับ แม้ตัวใดตัวหนึ่งจะล้มเหลวระหว่างทาง:

import contextlib

names = ["a", "b", "c"]
with contextlib.ExitStack() as stack:
    resources = [stack.enter_context(managed_resource(n)) for n in names]
    stack.callback(print, "extra cleanup callback ran")
    # use resources here
# everything registered with enter_context/callback unwinds here, most-recently-registered first

stack.enter_context(cm) เรียก cm.__enter__() ทันทีและลงทะเบียน cm.__exit__ ให้ทำงานเมื่อ ExitStack เองออก stack.callback(fn, *args) ลงทะเบียน function ธรรมดาให้ทำงานตอนเดียวกัน สำหรับ cleanup ที่ไม่ใช่ context manager เอง ทุกอย่างที่ลงทะเบียนไว้ unwind ตามลำดับ LIFO — ลงทะเบียนหลังสุด ทำงานก่อน — เหมือนกับที่ with statement ที่ซ้อนกัน unwind จากในสุดออกมา

การซ้อน context manager

context manager หลายตัวรวมกันในบรรทัด with เดียวได้ — พวกมันซ้อนจากซ้ายไปขวา:

with open("in.txt") as src, open("out.txt", "w") as dst:
    dst.write(src.read())

`async with` — ตัวอย่างสั้น ๆ

asynchronous context manager implement __aenter__ และ __aexit__ เป็น coroutine method แทน method ธรรมดา และใช้กับ async with ภายใน async def function:

class AsyncTimer:
    async def __aenter__(self) -> Self:
        print("async timer started")
        return self
    async def __aexit__(self, exc_type, exc, tb) -> bool:
        print("async timer stopped")
        return False

async def main() -> None:
    async with AsyncTimer():
        print("doing async work...")

protocol ที่เหลือเหมือนเดิมทุกประการกับเวอร์ชัน synchronous — argument สามตัวเหมือนกัน ความหมายของ return value เหมือนกัน โมดูล Async & Concurrency ครอบคลุม async/await, task และ event loop แบบลึกกว่านี้ ส่วนนี้มีไว้แค่ให้ จำรูปร่างได้เมื่อเจอเท่านั้น

Demo แบบ runnable เต็มรูปแบบ

Open in an online REPL

import contextlib
from collections.abc import Iterator
from types import TracebackType
from typing import Self

# --- class-based context manager ---
class Timer:
    def __enter__(self) -> Self:
        print("Timer started")
        return self
    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> bool:
        print("Timer stopped")
        suppress = False
        return suppress

print("=== class-based ===")
with Timer():
    print("  doing work...")

# --- contextlib.contextmanager ---
@contextlib.contextmanager
def managed_resource(name: str) -> Iterator[str]:
    print(f"Acquiring {name}")
    try:
        yield name
    finally:
        print(f"Releasing {name}")

print("\n=== contextmanager generator ===")
with managed_resource("connection") as res:
    print(f"  using {res}")

# --- cleanup on error ---
print("\n=== cleanup runs even on error ===")
try:
    with managed_resource("db") as res:
        print(f"  using {res}")
        raise RuntimeError("something failed")
except RuntimeError as e:
    print(f"  caught: {e}")

Depth demo: ทุกอย่างข้างบน รันจริง

Open in an online REPL

import asyncio
import contextlib
from collections.abc import Iterator
from types import TracebackType
from typing import Self

# --- __exit__ returning True swallows the exception ---
print("=== __exit__ returning True swallows ===")
class Swallower:
    def __enter__(self) -> Self:
        return self
    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> bool:
        if exc is not None:
            print(f"  swallowing {type(exc).__name__}: {exc}")
        return True  # BUG-PRONE if unintentional: any exception disappears here

with Swallower():
    raise ValueError("this never propagates")
print("  execution continues here -- the ValueError above is gone")

# --- contextlib.suppress is built on exactly this pattern ---
print("=== contextlib.suppress ===")
with contextlib.suppress(FileNotFoundError):
    open("does-not-exist.txt")
print("  suppress(FileNotFoundError).__exit__ returns True only for FileNotFoundError")

# --- ExitStack: a dynamic number of context managers ---
print("=== ExitStack ===")
@contextlib.contextmanager
def resource(name: str) -> Iterator[str]:
    print(f"  acquiring {name}")
    try:
        yield name
    finally:
        print(f"  releasing {name}")

names = ["a", "b", "c"]
with contextlib.ExitStack() as stack:
    resources = [stack.enter_context(resource(n)) for n in names]
    print(f"  using {resources}")
    stack.callback(print, "  extra cleanup callback ran")
# unwinds LIFO: the callback (registered last) runs first, then c, then b, then a

# --- async with teaser ---
print("=== async with (teaser) ===")
class AsyncTimer:
    async def __aenter__(self) -> Self:
        print("  async timer started")
        return self
    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc: BaseException | None,
        tb: TracebackType | None,
    ) -> bool:
        print("  async timer stopped")
        return False

async def main() -> None:
    async with AsyncTimer():
        print("  doing async work...")

asyncio.run(main())

Mental Model

context manager คือคำสัญญาสองส่วน: __enter__ ทำงานก่อน body เสมอ และ — ไม่ว่า body จะออกอย่างไร ปกติ, ผ่าน return, หรือผ่าน exception ใดก็ตาม — __exit__ ทำงานหลังจากนั้นเสมอ with statement คือรูปร่างเดียวกันกับ try/except/finally ที่แสดงไว้ตอนต้นบทเป๊ะ แค่สะกดต่างกัน จุดเดียวที่รูปร่างนี้ต่างจาก finally block ธรรมดา คือ return value ของ __exit__ ซึ่งสามารถเลือก suppress สิ่งที่ปกติจะ propagate ได้อย่างจริงจัง

ข้อแลกเปลี่ยน

สิ่งที่ได้	ประโยชน์	ต้นทุน
Class-based context manager	ควบคุมได้เต็มที่ เก็บ state ข้าม `__enter__`/`__exit__` ได้	boilerplate มากกว่า generator
`contextlib.contextmanager`	function เดียว `try`/`finally` อ่านเป็นธรรมชาติ	มี `yield` เดียว — re-enter ซ้ำแบบที่ class-based บางแบบทำได้ไม่ได้
`contextlib.suppress`	บรรทัดเดียว ระบุชัดว่าเมินอะไร	ใช้ได้แค่กรณีแคบ ๆ ของ “เมิน exception type ที่ระบุนี้”
`ExitStack`	จัดการ resource จำนวนไม่แน่นอนได้อย่างปลอดภัย	อ้อมกว่า `with a, b, c:` ตรง ๆ เล็กน้อย

ความเข้าใจผิดที่พบบ่อย

“__exit__ คืน False แปลว่ามีอะไรผิดพลาด” — False (หรือ None หรือไม่คืนอะไรเลย) คือ default ที่ ปกติ และถูกต้อง — มันแปลว่า “ปล่อยให้ exception propagate เหมือนที่มันจะเป็นถ้าไม่มี context manager นี้” การคืน True ต่างหากคือทางเลือกที่ผิดปกติที่ต้องมีเหตุผล
“with statement สำคัญแค่กับ exception เท่านั้น” — มันยังรับประกันว่า cleanup ทำงานตอน return ปกติจาก ข้างใน block หรือ break/continue ออกจาก loop ที่ครอบอยู่ด้วย — ไม่ว่า body จะออกทางไหนก็ตาม
“function ที่ decorate ด้วย contextlib.contextmanager ใช้ซ้ำได้ในหลาย with statement” — generator จะหมดอายุหลังจากรันจบรอบเดียว ต้องเรียก function ที่ decorate ไว้ใหม่สำหรับ with block ใหม่ แทนการใช้ generator object เดิมซ้ำ
“ExitStack มีประโยชน์แค่กับไฟล์” — มันใช้ได้กับ context manager ใดก็ได้ (lock, network connection, state ชั่วคราว) และกับ cleanup callback ธรรมดาที่ไม่ใช่ context manager เลยด้วย

อ่านโค้ดที่ AI เขียน

ก่อนเชื่อ context manager ที่ AI generate ให้ตรวจสิ่งเหล่านี้:

__exit__ คืน True (หรืออะไรที่ truthy ชัดเจน) โดยไม่มีเหตุผลที่ comment ไว้ชัดเจนหรือเปล่า? มันจะลบทุก exception ที่ raise ภายในทุก with block ที่ใช้มันแบบเงียบ ๆ
resource ถูกเปิดและปิดด้วยมือ (f = open(...) / f.close()) แทนที่จะเป็น with open(...) as f: หรือเปล่า? exception ใดก็ตามระหว่างสองการเรียกนี้จะข้าม close() ไป
@contextlib.contextmanager generator ทำ cleanup นอก try/finally หรือเปล่า? ถ้าโค้ดระหว่าง yield กับ cleanup raise ได้ หรือถ้า with body เอง raise cleanup ที่เขียนโดยไม่มี finally จะไม่ทำงานเลย
ใช้ with statement จำนวนคงที่ที่ hardcode ไว้ ทั้งที่จำนวน resource จริง ๆ ไม่แน่นอน (list ของไฟล์, จำนวน connection ที่แปรผัน) หรือเปล่า? นั่นมักแปลว่าต้องใช้ ExitStack
contextlib.suppress (หรือ bare except: pass) กำลังซ่อน exception ที่ควร log ไว้หรือเปล่า? การ suppress คือการตัดสินใจ ไม่ใช่ default

🐛 หา bug ในโค้ดนี้

class DatabaseTransaction:
  def __enter__(self):
      print("BEGIN")
      return self

  def __exit__(self, exc_type, exc, tb):
      if exc_type is not None:
          print("ROLLBACK")
      else:
          print("COMMIT")
      return True

with DatabaseTransaction():
  print("INSERT INTO users ...")
  raise ValueError("insert failed")

print("continuing after the transaction")

เฉลย

__exit__ คืน True แบบไม่มีเงื่อนไข — แม้แต่บน path ROLLBACK ที่มี exception เกิดขึ้นจริง นั่นแปลว่า ValueError("insert failed") ถูกกลืนไปแบบเงียบ ๆ: โค้ดพิมพ์ BEGIN, INSERT INTO users ..., ROLLBACK แล้วก็ continuing after the transaction โดยไม่มีร่องรอยว่ามีอะไรล้มเหลวเลย caller ไม่มีทางรู้ว่า insert ไม่สำเร็จ ทางแก้คือ return False (หรือไม่ต้อง return เลย) เพื่อให้ exception ยัง propagate ต่อหลัง rollback — __exit__ ควรตัดสินใจว่าจะรัน cleanup หรือไม่ ไม่ใช่ตัดสินใจว่าจะซ่อน failure หรือไม่

object ต้อง implement method อะไรสองตัวถึงจะใช้เป็น context manager ได้? __open__ และ __close__ __enter__ และ __exit__ __start__ และ __stop__ __setup__ และ __teardown__

ใน @contextlib.contextmanager generator อะไรเป็นเส้นแบ่งระหว่าง setup กับ teardown? return yield raise StopIteration break

method __exit__ ทำงานเมื่อไหร่? เฉพาะเมื่อ with block ออกแบบปกติเท่านั้น เฉพาะเมื่อมี exception ถูก raise ภายใน with block เท่านั้น เสมอ -- ไม่ว่า block จะออกแบบปกติหรือด้วย exception เฉพาะเมื่อ context manager ถูก garbage-collect เท่านั้น

โค้ดนี้พิมพ์อะไร?

class CM:
    def __enter__(self):
        return self
    def __exit__(self, exc_type, exc, tb):
        return True

with CM():
    raise ValueError("x")
print("after")

raise ValueError: x โดยไม่มีใครจับ 'after' -- ValueError ถูก suppress แบบเงียบ ๆ เพราะ __exit__ คืน True ไม่มีอะไรเลย -- โปรแกรม exit ทันที พิมพ์ 'after' แล้วค่อย re-raise ValueError

stack.enter_context(cm) ภายใน with contextlib.ExitStack() as stack: block ทำอะไร? แค่ลงทะเบียน cm ไว้สำหรับ cleanup ทีหลัง โดยไม่เรียก __enter__ เรียก cm.__enter__() ทันที และลงทะเบียน cm.__exit__ ให้ทำงานเมื่อ ExitStack ออก สร้าง ExitStack ที่ซ้อนกัน ต้องการให้ cm ถูก enter ไปแล้ว

เมื่อ with block ออก callback กับ cleanup ของ resource ทำงานตามลำดับไหน?

with contextlib.ExitStack() as stack:
    stack.callback(print, "first")
    stack.enter_context(managed_resource("r"))

callback ทำงานก่อน เพราะลงทะเบียนก่อน cleanup ของ resource ทำงานก่อน เพราะลงทะเบียนทีหลัง -- ExitStack unwind แบบ LIFO ทั้งคู่ทำงานพร้อมกัน ลำดับไม่แน่นอน

ทำไมต้องใช้ async def __aenter__ / async def __aexit__ แทน __enter__ / __exit__? ไม่เกี่ยวข้องกับ protocol ของ with statement เลย async with ต้องการให้ method setup/teardown เป็น coroutine ที่ถูก await เป็นส่วนหนึ่งของการเข้าและออกจาก block __aenter__ และ __aexit__ แทนที่ __enter__ และ __exit__ ทั้งหมดใน Python 3.14 จำเป็นแค่ตอนที่ resource เองทำ network I/O เท่านั้น