Контекстные менеджеры: with, contextlib, ExitStack

Когда работаешь с ресурсом - файлом, сетевым соединением, локом, транзакцией БД - нужно гарантировать освобождение даже при ошибках. Можно делать это через try/finally, но это многословно. Python даёт with-statement - декларативный способ управления ресурсами. Это последний урок Модуля 3.

Зачем нужен with

Без context manager:

f = open("data.txt")
try:
    data = f.read()
    process(data)
finally:
    f.close()  # всегда закрыть, даже при ошибке

С context manager:

with open("data.txt") as f:
    data = f.read()
    process(data)
# файл автоматически закрыт здесь

Это короче, безопаснее и идиоматичнее. with гарантирует вызов cleanup-кода независимо от того, что произошло внутри блока.

Множественные ресурсы

with open("input.txt") as fin, open("output.txt", "w") as fout:
    fout.write(fin.read().upper())

Через запятую можно перечислять несколько менеджеров. Они закроются в обратном порядке (LIFO) - сначала последний открытый, потом первый.

С Python 3.10 можно использовать скобки для многострочного варианта:

with (
    open("input1.txt") as f1,
    open("input2.txt") as f2,
    open("output.txt", "w") as out,
):
    out.write(f1.read() + f2.read())

Что может быть context manager

Стандартные:

open() - файлы
threading.Lock(), asyncio.Lock() - локи
socket.socket() - сокеты
sqlite3.connect() - подключение к БД
subprocess.Popen() - дочерние процессы
tempfile.NamedTemporaryFile() - временные файлы
contextlib.suppress() - подавление исключений
decimal.localcontext() - локальный контекст Decimal

Создаём свой - через класс

Context manager - это объект с dunder-методами __enter__ и __exit__:

class Timer:
    def __enter__(self):
        import time
        self.start = time.time()
        return self  # возвращаемое значение for `as`

    def __exit__(self, exc_type, exc_val, exc_tb):
        import time
        elapsed = time.time() - self.start
        print(f"Прошло: {elapsed:.3f}s")
        # return None - не подавлять исключения

Использование:

with Timer() as t:
    do_heavy_work()
# Прошло: 1.234s

__enter__ возвращает значение, которое привяжется к переменной после as. __exit__ принимает информацию об исключении (или три None если без ошибки) и опционально возвращает True чтобы подавить его.

Если `__exit__` возвращает True, исключение **подавляется** - оно не пробрасывается дальше. Это полезно для специфичного поведения (например, `contextlib.suppress`), но не делай это случайно - все ошибки молча проглатываются. По умолчанию возвращай None (то есть `return` ничего).

Создаём свой - через contextlib.contextmanager

Более удобный способ - декоратор @contextmanager:

from contextlib import contextmanager
import time

@contextmanager
def timer():
    start = time.time()
    try:
        yield  # сюда передастся управление в блок with
    finally:
        elapsed = time.time() - start
        print(f"Прошло: {elapsed:.3f}s")

with timer():
    do_heavy_work()

Структура (про сам yield - в уроке про генераторы):

До yield - setup (как __enter__)
yield value - возвращает значение для as
После yield - teardown (как __exit__)
try/finally гарантирует cleanup при исключении

Пример с возвращаемым значением:

@contextmanager
def opened_resource(name):
    print(f"Открываем {name}")
    resource = {"name": name, "data": []}
    try:
        yield resource
    finally:
        print(f"Закрываем {name}")

with opened_resource("DB") as res:
    res["data"].append("row")
    print(res)
# Открываем DB
# {'name': 'DB', 'data': ['row']}
# Закрываем DB

contextlib - готовые менеджеры

suppress - подавить указанные исключения:

from contextlib import suppress

with suppress(FileNotFoundError):
    os.remove("optional.txt")
# вместо try/except FileNotFoundError: pass

redirect_stdout/redirect_stderr - перенаправить вывод:

from contextlib import redirect_stdout
import io

buf = io.StringIO()
with redirect_stdout(buf):
    print("это уйдёт в buf")

print(buf.getvalue())  # "это уйдёт в buf\n"

closing - вызывает .close() для объектов без __exit__:

from contextlib import closing

with closing(urllib.urlopen("http://example.com")) as page:
    data = page.read()

ExitStack - динамические менеджеры

Когда количество ресурсов неизвестно заранее, используется ExitStack:

from contextlib import ExitStack

files = ["a.txt", "b.txt", "c.txt"]

with ExitStack() as stack:
    handles = [stack.enter_context(open(f)) for f in files]
    # все файлы открыты, обрабатываем
    for h in handles:
        process(h.read())
# все закрыты в обратном порядке

enter_context регистрирует менеджер в стеке. При выходе они закрываются в обратном порядке.

Также ExitStack поддерживает добавление callbacks:

with ExitStack() as stack:
    conn = open_db()
    stack.callback(conn.close)
    # ... используем conn
# conn.close() вызовется автоматически

Async context managers

С asyncio есть аналог async with:

async with aiohttp.ClientSession() as session:
    async with session.get(url) as response:
        data = await response.text()

Для своих async-менеджеров - __aenter__/__aexit__ или @asynccontextmanager:

from contextlib import asynccontextmanager

@asynccontextmanager
async def db_transaction(conn):
    tx = await conn.begin()
    try:
        yield tx
        await tx.commit()
    except Exception:
        await tx.rollback()
        raise

async with db_transaction(conn) as tx:
    await tx.execute("INSERT INTO ...")

Это будет важно в Модуле 7 при изучении asyncio.

Типичные применения

1. Файлы (самое частое):

with open("data.txt") as f:
    for line in f:
        process(line)

2. Блокировки:

import threading
lock = threading.Lock()

with lock:
    # критическая секция
    shared_resource += 1

3. Транзакции БД:

with connection.transaction():
    cursor.execute("INSERT ...")
    cursor.execute("UPDATE ...")
# COMMIT или ROLLBACK

4. Временное изменение окружения:

@contextmanager
def temp_env(key, value):
    old = os.environ.get(key)
    os.environ[key] = value
    try:
        yield
    finally:
        if old is None:
            del os.environ[key]
        else:
            os.environ[key] = old

with temp_env("DEBUG", "1"):
    run_with_debug()

5. Mocking в тестах:

from unittest.mock import patch

with patch("module.api_call") as mock_api:
    mock_api.return_value = {"ok": True}
    result = function_using_api()

Антипаттерны

1. Использование try/finally когда есть готовый менеджер.

# Плохо
f = open("file.txt")
try:
    data = f.read()
finally:
    f.close()

# Хорошо
with open("file.txt") as f:
    data = f.read()

2. Подавление всех исключений в exit.

class Bad:
    def __exit__(self, *args):
        return True   # ВСЕ ошибки молча проглатываются!

Это маскирует баги. Подавляй только конкретные ожидаемые исключения.

3. Игнорирование async with в async-коде.

# Плохо - блокирует event loop
async def fetch():
    with session.get(url) as resp:  # синхронно!
        ...

# Хорошо
async def fetch():
    async with session.get(url) as resp:
        ...

Сравнение с Go и PHP

Язык	Аналог `with`
Go	`defer` - cleanup откладывается до выхода из функции, в обратном порядке регистрации
PHP	Нет встроенного механизма, освобождение ресурсов делается вручную через try/finally или RAII-style классы

Python with декларативный: cleanup привязан к scope блока, не к выходу из функции (как Go defer). Это удобно когда хочется явно ограничить время жизни ресурса конкретным блоком кода, а не всей функцией.

Мини-задание

Реализуй timer как context manager:

from contextlib import contextmanager
import time

@contextmanager
def timer(label="block"):
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        print(f"{label}: {elapsed:.3f}s")

with timer("обработка"):
    sum(x ** 2 for x in range(10_000_000))

Многократное использование ExitStack:

from contextlib import ExitStack

def merge_files(input_paths, output_path):
    with ExitStack() as stack:
        inputs = [stack.enter_context(open(p)) for p in input_paths]
        output = stack.enter_context(open(output_path, "w"))
        for f in inputs:
            output.write(f.read())

Создай менеджер для временного изменения переменной:

@contextmanager
def temp_attr(obj, attr, value):
    old = getattr(obj, attr)
    setattr(obj, attr, value)
    try:
        yield
    finally:
        setattr(obj, attr, old)

# Использование
class Config:
    debug = False

config = Config()
with temp_attr(config, "debug", True):
    print(config.debug)  # True
print(config.debug)  # False

Что дальше

Модуль 3 завершён. Освоили условия, циклы, pattern matching, comprehensions, исключения и контекстные менеджеры - полный набор инструментов управления потоком. В следующем модуле перейдём к функциям: аргументы, замыкания, декораторы и продвинутые техники.