Context Managers – مدیریت منابع

Context Manager‌ها مکانیزم پایتون برای مدیریت منابع هستند – باز کردن فایل، اتصال دیتابیس، lock و هر چیزی که نیاز به cleanup دارد. در این فصل با with statement، پیاده‌سازی Context Manager سفارشی، ماژول contextlib و الگوهای پیشرفته آشنا می‌شویم.

چرا Context Manager؟

بدون context manager، کد ما باید cleanup را دستی انجام دهد:

# کد بد - cleanup دستی
f = open("data.txt")
try:
    data = f.read()
    process(data)
finally:
    f.close()  # حتی اگر خطا رخ دهد

# کد خوب - با with
with open("data.txt") as f:
    data = f.read()
    process(data)
# فایل خودکار بسته می‌شود

پروتکل Context Manager

یک کلاس برای context manager بودن، باید دو متد داشته باشد:

class MyContext:
    def __enter__(self):
        """شروع context - شیء قابل استفاده برمی‌گرداند"""
        print("ورود")
        return self  # یا هر چیز دیگر
    
    def __exit__(self, exc_type, exc_value, traceback):
        """پایان context - cleanup"""
        print("خروج")
        # return False = استثنا را propagate کن
        # return True  = استثنا را خاموش کن
        return False

with MyContext() as ctx:
    print("داخل")
# ورود
# داخل
# خروج

پارامترهای __exit__

• exc_type – نوع استثنا (None اگر بدون خطا)
• exc_value – شیء استثنا
• traceback – traceback object

پیاده‌سازی‌های کاربردی

۱. Timer

import time

class Timer:
    def __init__(self, name="Block"):
        self.name = name
    
    def __enter__(self):
        self.start = time.perf_counter()
        return self
    
    def __exit__(self, *exc):
        self.elapsed = time.perf_counter() - self.start
        print(f"{self.name}: {self.elapsed:.4f} ثانیه")

with Timer("محاسبه"):
    sum(range(10_000_000))
# محاسبه: 0.156 ثانیه

۲. اتصال دیتابیس

class DatabaseConnection:
    def __init__(self, host, db):
        self.host = host
        self.db = db
        self.conn = None
    
    def __enter__(self):
        import mysql.connector
        print(f"اتصال به {self.db}")
        self.conn = mysql.connector.connect(
            host=self.host,
            database=self.db,
            user="root", password=""
        )
        return self.conn.cursor()
    
    def __exit__(self, exc_type, exc_value, tb):
        if exc_type:
            print(f"خطا - rollback: {exc_value}")
            self.conn.rollback()
        else:
            print("commit")
            self.conn.commit()
        self.conn.close()
        return False  # raise مجدد استثنا

with DatabaseConnection("localhost", "shop") as cursor:
    cursor.execute("INSERT INTO products VALUES (...)")
    cursor.execute("UPDATE inventory SET ...")
# اگر خطا رخ دهد: rollback خودکار
# در غیر این صورت: commit خودکار

۳. تغییر موقت دایرکتوری

import os

class ChangeDir:
    def __init__(self, new_path):
        self.new_path = new_path
        self.original = None
    
    def __enter__(self):
        self.original = os.getcwd()
        os.chdir(self.new_path)
        return self.new_path
    
    def __exit__(self, *exc):
        os.chdir(self.original)

with ChangeDir("/tmp"):
    print(os.getcwd())  # /tmp
    # کارهایی در /tmp
print(os.getcwd())  # برمی‌گردد به مسیر اصلی

ماژول contextlib

@contextmanager – راه ساده‌تر

به‌جای پیاده‌سازی __enter__ و __exit__، از یک generator استفاده کنید:

from contextlib import contextmanager
import time

@contextmanager
def timer(name="Block"):
    start = time.perf_counter()
    try:
        yield  # اینجا کد داخل with اجرا می‌شود
    finally:
        elapsed = time.perf_counter() - start
        print(f"{name}: {elapsed:.4f}s")

with timer("محاسبه"):
    sum(range(10_000_000))

الگوی setup/teardown

@contextmanager
def database_transaction(conn):
    """transaction خودکار"""
    cursor = conn.cursor()
    try:
        yield cursor
        conn.commit()
    except Exception as e:
        conn.rollback()
        raise
    finally:
        cursor.close()

# استفاده
with database_transaction(conn) as cursor:
    cursor.execute("INSERT ...")
    cursor.execute("UPDATE ...")
    # اگر خطا → rollback خودکار

ابزارهای دیگر contextlib

suppress – خاموش کردن استثناهای خاص

from contextlib import suppress
import os

# روش قدیمی
try:
    os.remove("file.txt")
except FileNotFoundError:
    pass

# روش جدید با suppress
with suppress(FileNotFoundError):
    os.remove("file.txt")

# چند نوع استثنا
with suppress(FileNotFoundError, PermissionError):
    os.remove("file.txt")

redirect_stdout / redirect_stderr

from contextlib import redirect_stdout
import io

# گرفتن خروجی print
buffer = io.StringIO()
with redirect_stdout(buffer):
    print("سلام")
    print("دنیا")

content = buffer.getvalue()
print(repr(content))  # 'سلامnدنیاn'

# نوشتن به فایل
with open("output.log", "w") as f:
    with redirect_stdout(f):
        print("این به فایل می‌رود")
        print("نه ترمینال")

closing – بستن خودکار

from contextlib import closing
from urllib.request import urlopen

# urlopen خودش context manager است، اما برای اشیائی که نیستند:
with closing(urlopen("https://api.example.com")) as response:
    data = response.read()
# response.close() خودکار

ExitStack – چند context manager پویا

from contextlib import ExitStack

# باز کردن چند فایل به‌تعداد متغیر
filenames = ["a.txt", "b.txt", "c.txt"]

with ExitStack() as stack:
    files = [
        stack.enter_context(open(name, "r"))
        for name in filenames
    ]
    # کار با همه فایل‌ها
    for f in files:
        print(f.readline())
# همه فایل‌ها خودکار بسته می‌شوند

Context Manager‌های قابل استفاده مجدد

from contextlib import contextmanager

@contextmanager
def open_file(name, mode):
    f = open(name, mode)
    try:
        yield f
    finally:
        f.close()

cm = open_file("data.txt", "r")
# می‌توان چند بار استفاده کرد
with cm as f:
    pass
# اما در همان context تنها یک‌بار - باید دوباره ساخت

Async Context Managers

برای async (که در فصل بعد بررسی می‌کنیم):

import aiohttp
import asyncio

class AsyncDatabase:
    async def __aenter__(self):
        self.conn = await connect_async(...)
        return self.conn
    
    async def __aexit__(self, *exc):
        await self.conn.close()

# با @asynccontextmanager
from contextlib import asynccontextmanager

@asynccontextmanager
async def async_session():
    session = aiohttp.ClientSession()
    try:
        yield session
    finally:
        await session.close()

# استفاده
async def main():
    async with async_session() as session:
        async with session.get("https://api.example.com") as resp:
            data = await resp.json()

مدیریت استثنا در Context Manager

class IgnoreErrors:
    def __init__(self, *exceptions):
        self.exceptions = exceptions
    
    def __enter__(self):
        return self
    
    def __exit__(self, exc_type, exc_value, tb):
        if exc_type and issubclass(exc_type, self.exceptions):
            print(f"خطای {exc_type.__name__} نادیده گرفته شد")
            return True  # خاموش کردن استثنا
        return False  # propagate

with IgnoreErrors(ZeroDivisionError):
    x = 1 / 0
    print("این اجرا نمی‌شود")
# خطای ZeroDivisionError نادیده گرفته شد
print("ادامه برنامه")

مثال‌های واقعی

Lock موقت

import threading

lock = threading.Lock()

@contextmanager
def acquire_lock_with_timeout(lock, timeout):
    if not lock.acquire(timeout=timeout):
        raise TimeoutError("نتوانستم lock بگیرم")
    try:
        yield
    finally:
        lock.release()

with acquire_lock_with_timeout(lock, 5):
    # کار با منبع مشترک
    pass

Mock موقت

from contextlib import contextmanager

@contextmanager
def mock_attribute(obj, attr, mock_value):
    """جایگزین کردن موقت یک attribute"""
    original = getattr(obj, attr)
    setattr(obj, attr, mock_value)
    try:
        yield
    finally:
        setattr(obj, attr, original)

# مفید برای تست
class Service:
    api_url = "https://real-api.com"

with mock_attribute(Service, "api_url", "https://test-api.com"):
    print(Service.api_url)  # https://test-api.com
print(Service.api_url)      # https://real-api.com

بهترین شیوه‌ها

• برای cleanup منابع، همیشه context manager استفاده کنید
• ساده‌ترین راه: @contextmanager از contextlib
• در __exit__، فقط return True اگر می‌خواهید استثنا را خاموش کنید
• برای cleanup‌های پیچیده، try/finally داخل generator
• ExitStack برای تعداد متغیر context manager

جمع‌بندی

• Context Manager منابع را خودکار مدیریت می‌کند
• پروتکل: __enter__ و __exit__
• @contextmanager راه آسان با generator
• contextlib ابزارهای آماده: suppress، redirect_stdout، closing، ExitStack
• Async نسخه: __aenter__/__aexit__ یا @asynccontextmanager