Learning Platform
Урок 05.02 · 18 мин
Начальный
Context managerwith statementcontextlibResource management
Протокол контекст-менеджеров: asynccontextmanager и ExitStack Транзакции SQL: ACID и isolation levels

Где это уже было

Вы уже писали with open(path) as f: — это и есть контекст-менеджер. На этом уроке разберём, что под капотом, и научимся писать свои. Зачем junior DE это уметь: половина production-кода связана с ресурсами, которые надо обязательно закрыть — файлы, подключения к БД, транзакции, файловые блокировки. И контекст-менеджер — единственный по-настоящему надёжный способ это сделать.

Зачем вообще нужен with

Простой код без контекст-менеджера:

f = open("data.csv")
process(f)
f.close()

Выглядит безобидно, но содержит баг. Если process(f) бросит исключение, f.close() не выполнится — и файл останется открытым. На больших ETL-джобах это означает утечку файловых дескрипторов: процесс через несколько часов получает Too many open files и умирает.

«Грамотный» вариант через try / finally:

f = open("data.csv")
try:
    process(f)
finally:
    f.close()

Работает, но писать так каждый раз — больно. И в нём легко ошибиться: положить open внутрь try, забыть finally, перепутать вложенность. Поэтому в Python ввели специальный синтаксис — with:

with open("data.csv") as f:
    process(f)
# f закрыт здесь — гарантированно, даже если внутри было исключение

Это та же самая try / finally-конструкция, только короче и не позволяет случайно её сломать.

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

Что внутри: протокол __enter__ / __exit__

Любой класс, у которого определены два метода — __enter__ и __exit__ — можно использовать в with. Эти методы называются

dunder-методами
.

class LoudFile:
    def __init__(self, path: str):
        self.path = path
        self.file = None

    def __enter__(self):
        print(f"opening {self.path}")
        self.file = open(self.path)
        return self.file        # это значение попадёт в `as f`

    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f"closing {self.path}")
        self.file.close()
        return False             # False/None — пропустить исключение наружу

with LoudFile("data.csv") as f:
    print(f.read())

Что здесь происходит. Сначала вычисляется выражение справа от with — создаётся объект LoudFile. Затем у этого объекта Python вызывает __enter__() — его возвращаемое значение попадает в переменную после as. После выполнения тела блока (нормально или через исключение) Python обязательно вызывает __exit__(exc_type, exc_val, exc_tb).

Три аргумента в __exit__ — это информация о произошедшем исключении. Если код в with отработал без ошибок, все три равны None. Если было исключение — это его тип, значение и traceback. Возвращаемое значение __exit__:

  • False или None — исключение пробрасывается дальше (типичный случай).
  • True — исключение гасится, как будто его не было. Используйте редко: чаще всего проглатывание ошибок — баг.
Жизненный цикл with-блока

with гарантирует, что __exit__ вызовется в обоих случаях — и при нормальном завершении, и при исключении

входmanager.__enter__()Открывает ресурс, возвращает то, что попадёт в `as ...`
тело блокаваш код внутри with
ветка А: ОКблок дочитан до конца
__exit__(None, None, None)закрытие ресурса
ветка Б: exceptionвнутри блока исключение
__exit__(exc_type, exc_val, exc_tb)закрытие + решение, пробрасывать или гаситьЕсли __exit__ вернёт True, исключение проглотится. Обычно возвращают False.

Этот протокол — единственное, что отличает обычный объект от контекст-менеджера. Всё остальное в with — синтаксический сахар над двумя dunder-методами.

@contextmanager: писать CM-функцию вместо класса

Класс на 15 строк ради двух методов — это много. В stdlib есть модуль contextlib, и его декоратор @contextmanager позволяет описать контекст-менеджер обычной функцией с одним yield.

from contextlib import contextmanager

@contextmanager
def loud_file(path: str):
    print(f"opening {path}")
    f = open(path)
    try:
        yield f                  # тут будет выполняться тело with
    finally:
        print(f"closing {path}")
        f.close()

with loud_file("data.csv") as f:
    print(f.read())

Правила игры. Всё до yield — это аналог __enter__. То, что отдано через yield — это значение для as. Всё после yield — это аналог __exit__. И главное правило: yield нужно обернуть в try / finally, иначе при исключении внутри with cleanup не отработает.

Почему DE этим пользуется чаще, чем классами: для одноразовой обвязки ресурса (открыть, отдать, закрыть) функция на 5 строк читается в разы лучше класса. Класс уместен, если CM нужно переиспользовать с разными параметрами и хранить какое-то сложное состояние.

Полезные утилиты из contextlib

В этом же модуле есть несколько готовых маленьких инструментов, которые экономят время.

contextlib.suppress — подавить конкретное исключение, не писать try / except / pass:

from contextlib import suppress

with suppress(FileNotFoundError):
    Path("temp.log").unlink()   # если файла нет — молча идём дальше

Это короче и читабельнее, чем try / except FileNotFoundError: pass. Главное — указывайте конкретный тип, не пишите suppress(Exception): голое подавление ошибок маскирует баги.

contextlib.closing — обернуть любой объект с методом .close() в контекст-менеджер. Полезно для старых API без поддержки with:

from contextlib import closing
from urllib.request import urlopen

with closing(urlopen("https://example.com")) as response:
    data = response.read()

contextlib.ExitStack — динамически собирать несколько CM в стек. Например, если нужно открыть N файлов, число которых известно только в runtime:

from contextlib import ExitStack

with ExitStack() as stack:
    files = [stack.enter_context(open(p)) for p in paths]
    # все файлы открыты, любая ошибка корректно закроет всё открытое

Несколько менеджеров в одной строке

Часто нужно одновременно открыть несколько ресурсов. Можно вкладывать with:

with open("in.csv") as src:
    with open("out.csv", "w") as dst:
        dst.write(src.read())

А можно перечислить через запятую — это эквивалентная запись, читается чище:

with open("in.csv") as src, open("out.csv", "w") as dst:
    dst.write(src.read())

С Python 3.10+ можно ещё и переносить на несколько строк через скобки (PEP 617):

with (
    open("in.csv") as src,
    open("out.csv", "w") as dst,
    timing("copy"),
):
    dst.write(src.read())

Это особенно полезно, когда менеджеров много, как в типичном ETL: входной файл + выходной файл + транзакция БД + замер времени.

DE-кейс: транзакция в БД

Большинство DB-клиентов в Python поддерживают with для транзакций. Идиоматичный паттерн через SQLAlchemy:

from sqlalchemy import create_engine, text

engine = create_engine("postgresql+psycopg://localhost/orders")

with engine.begin() as conn:
    conn.execute(text("INSERT INTO orders (id, total) VALUES (1, 100)"))
    conn.execute(text("INSERT INTO order_items (order_id, sku) VALUES (1, 'A')"))
# на выходе из with — COMMIT, если исключений не было; ROLLBACK, если было

Это тот самый «гарантированный cleanup»: если посередине что-то упало, изменения откатятся, и БД останется в согласованном состоянии. Без with пришлось бы писать try / except / rollback / raise с шансом забыть что-то.

DE-кейс: замер времени блока

Простой профайлер — иногда нужнее любых дашбордов. Напишем CM-функцию, которая печатает, сколько времени ушло на блок:

import time
from contextlib import contextmanager
from collections.abc import Iterator

@contextmanager
def timing(label: str) -> Iterator[None]:
    start = time.perf_counter()
    try:
        yield
    finally:
        elapsed = time.perf_counter() - start
        print(f"{label}: {elapsed:.3f}s")

with timing("read csv"):
    rows = list(read_csv("events.csv"))

with timing("filter and aggregate"):
    result = aggregate(rows)

В отличие от time.time(), мы используем time.perf_counter() — он монотонный и имеет максимальное разрешение, что нужно для замеров производительности.

DE-кейс: подключение к SQLite

Соберём то, что выучили, в один пример — обёртка над SQLite с автоматическим закрытием и откатом при ошибке:

import sqlite3
from contextlib import contextmanager
from collections.abc import Iterator

@contextmanager
def sqlite_tx(path: str) -> Iterator[sqlite3.Connection]:
    conn = sqlite3.connect(path)
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

with sqlite_tx("orders.db") as conn:
    conn.execute("INSERT INTO orders (id, total) VALUES (?, ?)", (1, 100))
    conn.execute("INSERT INTO orders (id, total) VALUES (?, ?)", (2, 200))
# COMMIT при успехе, ROLLBACK при исключении, в любом случае close()

Это production-grade паттерн. Три гарантии в одной обвязке: транзакция целиком (atomicity), нет утечки соединений, ошибки пробрасываются наружу.

Подводные камни

Не глотайте ошибки. __exit__, возвращающий True, превращает баг в «всё в порядке». Если действительно нужно подавить — делайте это явно через contextlib.suppress(SpecificError).

Не делайте ресурсо-захват вне with. Если открываете файл/соединение — сразу заворачивайте в with. Промежуточное состояние «уже открыто, но ещё не в with» — частый источник утечек.

Не путайте with и декораторы. with — оборачивает блок кода. Декоратор — оборачивает функцию. Это разные инструменты, не делайте через декораторы то, что естественно ложится на with (про декораторы будет следующий урок).

Не злоупотребляйте. Не каждое «надо что-то сделать в начале и в конце» — это контекст-менеджер. Если cleanup не критичен и блок маленький — обычная функция читается лучше.

Что мы получили

  • with — синтаксический сахар над try / finally, гарантирует cleanup при любом завершении.
  • Объект-CM реализует __enter__ и __exit__ — это весь протокол.
  • @contextmanager превращает функцию с одним yield в контекст-менеджер — обычно это короче и понятнее, чем класс.
  • contextlib.suppress, contextlib.closing, contextlib.ExitStack — три инструмента, которые покроют 90% бытовых задач.
  • DE-применение: транзакции в БД, замеры времени, временные файлы, file locks — везде, где есть пара «открыть/закрыть».

В следующем уроке — декораторы. Они принимают функцию и возвращают новую функцию, и это второй главный инструмент идиоматичного Python. Мы напишем @retry для нестабильного API и @timeit для профилирования.

Закончили урок?

Отметьте его как пройденный, чтобы отслеживать свой прогресс

Войдите чтобы оценить урок

Прогресс модуля
0 из 6