ข้ามไปยังเนื้อหา

Context Managers

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

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

with open("data.txt") as f:
content = f.read()
# f ถูกปิดเสมอที่นี่ แม้ read() จะ raise

Object ที่มี method __enter__ และ __exit__ คือ context manager

with expr as var:
body

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

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

__enter__ ตั้งค่า resource และคืนค่าที่ bind ให้ var __exit__ ทำงานเสมอเมื่อออกจาก block ถ้าคืนค่า truthy จะ suppress exception ถ้าคืน False (หรือ None) exception จะ propagate ต่อ

class Timer:
def __enter__(self) -> "Timer":
print("Timer started")
return self
def __exit__(self, exc_type, exc_val, exc_tb) -> bool:
print("Timer stopped")
return False # ไม่ suppress exceptions

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

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

import contextlib
@contextlib.contextmanager
def managed_resource(name: str):
print(f"Acquiring {name}")
try:
yield name # ค่าที่ bind ให้ as-variable
finally:
print(f"Releasing {name}")

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

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

with open("in.txt") as src, open("out.txt", "w") as dst:
dst.write(src.read())
import contextlib
# --- class-based context manager ---
class Timer:
def __enter__(self):
print("Timer started")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
print("Timer stopped")
return False
print("=== class-based ===")
with Timer():
print(" doing work...")
# --- contextlib.contextmanager ---
@contextlib.contextmanager
def managed_resource(name: 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}")
Object ต้อง implement method อะไรสองตัวเพื่อใช้เป็น context manager?
ใน generator ของ `@contextlib.contextmanager` อะไรเป็นตัวแบ่งระหว่าง setup และ teardown?
Method `__exit__` ทำงานเมื่อใด?
เกิดอะไรขึ้นถ้า `__exit__` คืนค่า truthy เมื่อมี exception เกิดขึ้น?