Context Managers — Python المتقدم

مدراء السياق (Context Managers)

تعرف with open(...) منذ دورة هياكل البيانات. with ليست خاصة بالملفات: إنها طريقة عامة لضمان تنفيذ خطوة التنظيف (إغلاق، تحرير، استعادة) مهما حدث، حتى لو رُمي خطأ في المنتصف.

المشكلة التي يحلها

f = open("data.txt", encoding="utf-8")
process(f)        # لو رمت خطأ...
f.close()         # ...لن يُنفَّذ هذا السطر

الحل اليدوي try/finally يعمل لكنه مطوّل ومتكرر؛ with تغلفه.

البروتوكول

أي كائن فيه:

- __enter__(self): يُنفَّذ عند الدخول، وما يرجعه يُسند للمتغير بعد as.

- __exit__(self, exc_type, exc_val, exc_tb): يُنفَّذ عند الخروج دائماً. إذا حدث خطأ تصله تفاصيله، وإلا تكون المعاملات الثلاثة None.

إرجاع True من __exit__ يبتلع الخطأ؛ ارجع False (أو لا شيء) إلا إذا كنت متأكداً أنك تريد إخفاءه.

الطريقة الأسهل: @contextmanager

from contextlib import contextmanager

@contextmanager
def opened_db(name):
    conn = connect(name)        # قبل yield = __enter__
    try:
        yield conn               # القيمة بعد as
    finally:
        conn.close()             # بعد yield = __exit__

try/finally حول yield ضرورية، وإلا لن يُنفَّذ التنظيف عند حدوث خطأ.

أدوات جاهزة في contextlib

| الأداة | الاستخدام |

|---|---|

| suppress(ErrorType) | تجاهل خطأ محدد متوقع |

| redirect_stdout(f) | توجيه print إلى ملف أو StringIO |

| ExitStack | عدد متغير من مدراء السياق |

| tempfile.TemporaryDirectory() | مجلد مؤقت يُحذف تلقائياً |

أمثلة استخدام شائعة

- الملفات والاتصالات بقواعد البيانات والشبكة.

- الأقفال في البرمجة المتوازية: with lock:.

- قياس زمن جزء من الكود.

- تغيير إعداد مؤقتاً ثم استعادته.

عدة مدراء معاً

with open("a.txt", encoding="utf-8") as src, open("b.txt", "w", encoding="utf-8") as dst:
    dst.write(src.read())

تمرين

اكتب مدير سياق timer(label) بـ @contextmanager يطبع زمن الكتلة، ومدير سياق working_directory(path) يغيّر المجلد الحالي مؤقتاً ثم يعيده (os.chdir) حتى لو حدث خطأ.

# استخدام with المدمج (encoding ضروري للنص العربي)
with open("test.txt", "w", encoding="utf-8") as f:
    f.write("مرحباً!")
# الملف مغلق تلقائياً هنا

# إنشاء Context Manager بالـ class
import time

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

    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f"الوقت: {time.perf_counter() - self.start:.4f} ثانية")
        return False  # لا نبتلع الأخطاء

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

# Context Manager مع معالجة الأخطاء
class DatabaseConnection:
    def __init__(self, host):
        self.host = host
        self.connection = None

    def __enter__(self):
        print(f"الاتصال بـ {self.host}...")
        self.connection = {"connected": True}
        return self.connection

    def __exit__(self, exc_type, exc_val, exc_tb):
        print("إغلاق الاتصال...")
        self.connection = None
        if exc_type is not None:
            print(f"حدث خطأ: {exc_val}")
        return False

with DatabaseConnection("localhost") as db:
    print("استعلام قاعدة البيانات...")

# باستخدام contextmanager decorator
from contextlib import contextmanager, suppress
import shutil
import tempfile

@contextmanager
def temp_directory():
    dirpath = tempfile.mkdtemp()
    print(f"إنشاء مجلد مؤقت: {dirpath}")
    try:
        yield dirpath
    finally:
        shutil.rmtree(dirpath)
        print("حذف المجلد المؤقت")

with temp_directory() as tmpdir:
    print(f"العمل في: {tmpdir}")

# عدة Context Managers في سطر واحد
with open("test.txt", encoding="utf-8") as fin, open("out.txt", "w", encoding="utf-8") as fout:
    fout.write(fin.read() + " (نسخة)")

# suppress لتجاهل خطأ محدد ومتوقع
with suppress(FileNotFoundError):
    with open("nonexistent.txt", encoding="utf-8") as f:
        print(f.read())

# تنظيف ملفات المثال
import os
os.remove("test.txt")
os.remove("out.txt")