6. Что такое контекстный менеджер в Python?

6. Что такое контекстный менеджер в Python?.md

Контекстные менеджеры в Python

Контекстный менеджер — объект, который управляет входом и выходом из блока with, гарантируя выполнение инициализации и очистки ресурсов независимо от того, возникло исключение или нет.

Реализуется через методы __enter__ / __exit__ или декоратор @contextmanager.

Классическая реализация

class FileManager:
    def __init__(self, filename, mode):
        self.filename = filename
        self.mode = mode
        self.file = None

    def __enter__(self):
        self.file = open(self.filename, self.mode)
        return self.file

    def __exit__(self, exc_type, exc_value, traceback):
        if self.file:
            self.file.close()

with FileManager("test.txt", "w") as f:
    f.write("Hello, world!")

Через contextlib (чаще на практике)

from contextlib import contextmanager

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

with open_file("test.txt", "w") as f:
    f.write("Hello")

Когда применяется

Везде, где нужно гарантированно освободить ресурс или откатить состояние:

  • Файлы (open)
  • Подключения к БД (commit / rollback)
  • Сетевые соединения
  • Локи (threading.Lock)
  • Транзакции

Преимущества: код безопаснее, читаемость выше, исключения обрабатываются корректно.

В Django

Транзакции

from django.db import transaction

# Автоматический rollback при исключении
with transaction.atomic():
    order = Order.objects.create(user=user, total=500)
    Payment.objects.create(order=order, amount=500)
    # Если здесь упадёт — оба объекта не сохранятся

Отключение сигналов (например, при импорте данных)

from contextlib import contextmanager
from django.db.models.signals import post_save

@contextmanager
def mute_signals(*signals):
    handlers = {}
    for signal in signals:
        handlers[signal] = signal.receivers
        signal.receivers = []
    try:
        yield
    finally:
        for signal, receivers in handlers.items():
            signal.receivers = receivers

with mute_signals(post_save):
    bulk_import_users(data)  # без лишних сигналов

Переключение базы данных

from django.test.utils import override_settings

with override_settings(DATABASES=test_db_config):
    run_migration_test()

В FastAPI

Lifespan — инициализация при старте приложения

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: подключаемся к БД, кешу и т.д.
    await database.connect()
    await redis.initialize()
    yield
    # Shutdown: освобождаем ресурсы
    await database.disconnect()
    await redis.close()

app = FastAPI(lifespan=lifespan)

Dependency Injection с очисткой (через yield)

from sqlalchemy.orm import Session

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/users/{user_id}")
def get_user(user_id: int, db: Session = Depends(get_db)):
    return db.query(User).filter(User.id == user_id).first()

FastAPI автоматически вызывает всё после yield после завершения запроса — даже если он упал с ошибкой.