โพรโทคอล __enter__ และ __exit__

โพรโทคอลตัวจัดการบริบท

class Managed:
    def __enter__(self):
        print("entering")
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        print("exiting")
        return False

with Managed() as m:
    print("inside")

ค่าที่ส่งคืนโดย __enter__

class Resource:
    def __enter__(self):
        self.value = 42
        return self  # bound to 'r'

    def __exit__(self, *args):
        self.value = None
        return False

with Resource() as r:
    print(r.value)  # 42

อาร์กิวเมนต์ของ __exit__

class Safe:
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is ValueError:
            print(f"Suppressed ValueError: {exc_val}")
            return True   # suppress the exception
        return False      # re-raise anything else

with Safe():
    raise ValueError("oops")  # suppressed

การระงับข้อยกเว้น

class Ignore:
    def __enter__(self): return self
    def __exit__(self, *args): return True  # swallow all exceptions

with Ignore():
    1 / 0

print("continues here")

ตัวอย่างการเชื่อมต่อฐานข้อมูล

class DBConn:
    def __enter__(self):
        self.conn = connect_db()
        return self.conn

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type:
            self.conn.rollback()
        else:
            self.conn.commit()
        self.conn.close()
        return False

ตัวจัดการบริบท Timer

import time

class Timer:
    def __enter__(self):
        self.start = time.perf_counter()
        return self

    def __exit__(self, *args):
        self.elapsed = time.perf_counter() - self.start
        print(f"Elapsed: {self.elapsed:.3f}s")
        return False

with Timer():
    sum(range(1_000_000))

การซ้อนตัวจัดการบริบท

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

ตัวจัดการที่ใช้ซ้ำได้กับตัวจัดการที่ใช้ครั้งเดียว

class Reusable:
    def __enter__(self):
        return []
    def __exit__(self, *a):
        return False

for _ in range(3):
    with Reusable() as lst:
        lst.append(1)
        print(lst)  # [1] each time

__aenter__ และ __aexit__ (แบบอะซิงก์)

class AsyncDB:
    async def __aenter__(self):
        self.conn = await connect()
        return self.conn

    async def __aexit__(self, *args):
        await self.conn.close()

# async with AsyncDB() as conn:
#     ...

การครอบอ็อบเจ็กต์ไฟล์

with open("data.txt", "r") as f:
    data = f.read()
# file is closed here automatically
print(f.closed)  # True

การใช้งาน Lock

import threading

lock = threading.Lock()

def update_shared():
    with lock:
        # critical section
        pass

ตรวจสอบอย่างรวดเร็ว

สรุป