بروتوكول __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

مدير سياق لمؤقت

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

تنفيذ قفل

import threading

lock = threading.Lock()

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

تحقق سريع

مراجعة