Где это уже было
Вы уже писали 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. Эти методы называются
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 гарантирует, что __exit__ вызовется в обоих случаях — и при нормальном завершении, и при исключении
Этот протокол — единственное, что отличает обычный объект от контекст-менеджера. Всё остальное в 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 для профилирования.