Python · Синтаксис · Продвинутый уровень

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

10 задач

Объекты, управляющие настройкой и очисткой через оператор `with`. Охватывает `__enter__`/`__exit__`, `contextlib.contextmanager` и паттерны управления ресурсами.

__enter__, __exit__ и обработка исключений детально

#
**Зачем нужны контекстные менеджеры** Управление ресурсами следует строгому паттерну: *получить → использовать → освободить*. Шаг освобождения должен происходить даже если во время 'использования' возникнет исключение. Без гарантированной очистки файлы остаются открытыми, соединения с БД текут, а блокировки никогда не снимаются. ```python # Повторяющийся код — каждый вызывающий должен писать этот шаблон f = open('data.txt') try: content = f.read() finally: f.close() # нужно закрыть даже если f.read() поднял исключение ``` Контекстный менеджер инкапсулирует настройку и очистку в один объект, а оператор `with` вызывает их автоматически: ```python with open('data.txt') as f: content = f.read() # файл закрывается здесь, даже если f.read() поднял исключение ``` **Протокол: `__enter__` и `__exit__`** ```python class Timer: def __enter__(self): import time self._start = time.perf_counter() return self # становится целью 'as' def __exit__(self, exc_type, exc_val, exc_tb): import time elapsed = time.perf_counter() - self._start print(f'elapsed: {elapsed:.4f}s') return False # False = не подавлять исключения with Timer() as t: sum(range(1_000_000)) # elapsed: 0.0312s ``` **Что означают три аргумента `__exit__`** `__exit__(self, exc_type, exc_val, exc_tb)` вызывается при завершении блока `with` — нормальном или из-за исключения. - Если блок завершился *без* исключения: все три — `None` - Если исключение *возникло*: `exc_type` — класс исключения, `exc_val` — экземпляр, `exc_tb` — traceback Возвращаемое значение контролирует распространение исключения: ```python class SuppressKeyError: def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): if exc_type is KeyError: print(f'Подавлен отсутствующий ключ: {exc_val}') return True # True = поглотить исключение return False # False = дать распространиться d = {'a': 1} with SuppressKeyError(): print(d['missing']) # Подавлен отсутствующий ключ: 'missing' print('continues here') # выполнение продолжается ``` Возврат `True` из `__exit__` поглощает исключение. `False` или `None` — позволяет распространиться. Подавляйте исключения только намеренно.

@contextmanager, suppress, ExitStack и вложение

#
**`@contextmanager` — контекстный менеджер как генератор** Писать полный класс для простых контекстных менеджеров многословно. `contextlib.contextmanager` позволяет написать его как функцию-генератор с одним `yield`: ```python from contextlib import contextmanager @contextmanager def managed_connection(host): conn = connect(host) # __enter__: настройка try: yield conn # цель 'as'; код блока 'with' выполняется здесь finally: conn.close() # __exit__: очистка — всегда выполняется with managed_connection('localhost') as conn: conn.query('SELECT 1') ``` `try/finally` вокруг `yield` обязателен. Без него исключение в блоке `with` пропустит `conn.close()` — что сводит на нет весь смысл: ```python @contextmanager def bad_manager(): # БАГ: отсутствует try/finally resource = acquire() yield resource release(resource) # никогда не достигается если тело бросает исключение! @contextmanager def good_manager(): # правильно resource = acquire() try: yield resource finally: release(resource) ``` **`contextlib.suppress(*exc_types)` — подавление исключений одной строкой** ```python from contextlib import suppress import os with suppress(FileNotFoundError): os.remove('maybe_exists.txt') ``` **`contextlib.ExitStack` — динамическая композиция контекстных менеджеров** `ExitStack` управляет переменным числом контекстных менеджеров: ```python from contextlib import ExitStack # Открыть переменное число файлов безопасно: filenames = ['a.txt', 'b.txt', 'c.txt'] with ExitStack() as stack: files = [stack.enter_context(open(f)) for f in filenames] for f in files: print(f.read()) # Все файлы закрыты здесь — даже если один read() бросил исключение # ExitStack как паттерн очистки при сбое: def setup_resources(): stack = ExitStack() try: conn = stack.enter_context(get_connection()) lock = stack.enter_context(get_lock()) return conn, lock, stack except Exception: stack.close() # очищаем всё что уже получили raise ``` **Вложение контекстных менеджеров** ```python with open('in.txt') as src, open('out.txt', 'w') as dst: dst.write(src.read()) ``` Входят слева направо, выходят справа налево (как стек).

класс vs @contextmanager, реальные паттерны, nullcontext, ошибки

#
**Класс vs `@contextmanager` — когда что выбрать** | Ситуация | Использовать | |---|---| | Простая настройка + очистка | `@contextmanager` | | Нужно подкласировать или добавить методы | Класс | | Хотите мокировать или подкласировать в тестах | Класс | | Сложная логика `__enter__` | Класс | **Реальные паттерны** ```python from contextlib import contextmanager # 1. Временная директория @contextmanager def temp_dir(): import tempfile, shutil d = tempfile.mkdtemp() try: yield d finally: shutil.rmtree(d) # 2. Временная замена атрибута @contextmanager def set_attr(obj, name, value): old = getattr(obj, name, None) setattr(obj, name, value) try: yield finally: if old is None: delattr(obj, name) else: setattr(obj, name, old) ``` **`contextlib.nullcontext` — контекстный менеджер-заглушка** Когда иногда нужен контекстный менеджер, а иногда нет: ```python from contextlib import nullcontext def process(data, lock=None): cm = lock if lock is not None else nullcontext() with cm: return expensive_computation(data) process(data, lock=threading.Lock()) # потокобезопасно process(data) # без блокировки ``` **Распространённые ошибки** Отсутствующий `try/finally` в `@contextmanager` — самая частая ошибка. Код очистки после `yield` пропускается если тело `with` бросает исключение. Случайный возврат значения из `__exit__` — любое истинное значение подавляет исключение. `return exc_tb` или `return 1` молча поглотят исключения. Использование контекстного менеджера после выхода из `with` — ресурс закрыт; переменная `as` ещё указывает на объект но он в закрытом состоянии.
01

Контекстный менеджер Timer

#

Напишите контекстный менеджер на основе класса `Timer`, который измеряет время выполнения. `__enter__` должен записать время начала и вернуть `self`. `__exit__` должен вычислить и сохранить время выполнения в `self.elapsed`. После блока `with`, `timer.elapsed` должен содержать время в секундах.

import time

class Timer:
    def __enter__(self):
        pass

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass


with Timer() as t:
    total = sum(range(1_000_000))

print(f'Elapsed: {t.elapsed:.4f}s')
Решение
import time

class Timer:
    def __enter__(self):
        self._start = time.time()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.elapsed = time.time() - self._start
        return False  # не подавляем исключение


with Timer() as t:
    total = sum(range(1_000_000))

print(f'Elapsed: {t.elapsed:.4f}s')
02

Управляемый менеджер файлов

#

Напишите контекстный менеджер `ManagedFile(path, mode)` на основе класса, открывающий файл в `__enter__` и закрывающий в `__exit__`. Если внутри блока `with` возникнет исключение, файл всё равно должен быть закрыт. Верните объект файла из `__enter__`.

class ManagedFile:
    def __init__(self, path, mode='r'):
        pass

    def __enter__(self):
        pass

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass


with ManagedFile('/tmp/test.txt', 'w') as f:
    f.write('привет контекстный менеджер\n')

with ManagedFile('/tmp/test.txt', 'r') as f:
    print(f.read())  # привет контекстный менеджер
Решение
class ManagedFile:
    def __init__(self, path, mode='r'):
        self.path = path
        self.mode = mode
        self._file = None

    def __enter__(self):
        self._file = open(self.path, self.mode)
        return self._file

    def __exit__(self, exc_type, exc_val, exc_tb):
        if self._file:
            self._file.close()
        return False


with ManagedFile('/tmp/test.txt', 'w') as f:
    f.write('привет контекстный менеджер\n')

with ManagedFile('/tmp/test.txt', 'r') as f:
    print(f.read())
03

Декоратор @contextmanager

#

Перепишите контекстный менеджер `Timer` из упражнения 1 с помощью `@contextlib.contextmanager` и функции-генератора вместо класса. Функция должна `yield`-нуть словарь `{'elapsed': None}` и обновить его временем выполнения после `yield`.

import time
from contextlib import contextmanager

@contextmanager
def timer():
    pass


with timer() as t:
    total = sum(range(1_000_000))

print(f"Elapsed: {t['elapsed']:.4f}s")
Решение
import time
from contextlib import contextmanager

@contextmanager
def timer():
    info = {'elapsed': None}
    start = time.time()
    try:
        yield info           # 'as' получает этот объект
    finally:
        info['elapsed'] = time.time() - start


with timer() as t:
    total = sum(range(1_000_000))

print(f"Elapsed: {t['elapsed']:.4f}s")
04

Контекстный менеджер suppress_and_log

#

Напишите контекстный менеджер `suppress_and_log(*exception_types)`, который подавляет любой из указанных типов исключений и печатает `'Suppressed: <сообщение исключения>'`. Другие исключения должны распространяться нормально.

from contextlib import contextmanager

@contextmanager
def suppress_and_log(*exception_types):
    pass


with suppress_and_log(ValueError, ZeroDivisionError):
    result = 1 / 0
    print('эта строка не будет достигнута')

print('выполнение продолжается после блока with')
Решение
from contextlib import contextmanager

@contextmanager
def suppress_and_log(*exception_types):
    try:
        yield
    except exception_types as e:
        print(f'Suppressed: {e}')


with suppress_and_log(ValueError, ZeroDivisionError):
    result = 1 / 0
    print('эта строка не будет достигнута')

print('выполнение продолжается после блока with')
# Suppressed: division by zero
# выполнение продолжается после блока with
05

Контекстный менеджер временного файла

#

Напишите контекстный менеджер `temp_file(suffix='.txt')` через `@contextmanager`, создающий временный файл, возвращающий его путь и удаляющий файл при выходе из блока `with` (даже при исключении). Используйте `tempfile.mktemp()` для генерации пути.

import os
import tempfile
from contextlib import contextmanager

@contextmanager
def temp_file(suffix='.txt'):
    pass


with temp_file() as path:
    with open(path, 'w') as f:
        f.write('временные данные')
    print(os.path.exists(path))  # True

print(os.path.exists(path))  # False — файл удалён
Решение
import os
import tempfile
from contextlib import contextmanager

@contextmanager
def temp_file(suffix='.txt'):
    path = tempfile.mktemp(suffix=suffix)
    try:
        yield path
    finally:
        if os.path.exists(path):
            os.remove(path)


with temp_file() as path:
    with open(path, 'w') as f:
        f.write('временные данные')
    print(os.path.exists(path))  # True

print(os.path.exists(path))  # False
06

Контекстный менеджер с отступом вывода

#

Напишите контекстный менеджер `indented(level=1, indent=' ')`, который делает все вызовы `print()` внутри блока `with` с дополнительным отступом. Временно заменяйте встроенную функцию `print` через `builtins.print`. После выхода из блока восстановите оригинальный `print`.

import builtins
from contextlib import contextmanager

@contextmanager
def indented(level=1, indent='  '):
    pass


print('верхний уровень')
with indented(2):
    print('отступ 4 пробела')
    print('тоже отступ')
print('возврат на верхний уровень')
Решение
import builtins
from contextlib import contextmanager

@contextmanager
def indented(level=1, indent='  '):
    prefix = indent * level
    original_print = builtins.print
    def indented_print(*args, **kwargs):
        original_print(prefix, end='')
        original_print(*args, **kwargs)
    builtins.print = indented_print
    try:
        yield
    finally:
        builtins.print = original_print


print('верхний уровень')
with indented(2):
    print('отступ 4 пробела')
    print('тоже отступ')
print('возврат на верхний уровень')
07

Контекстный менеджер транзакций

#

Напишите класс `FakeDB` со списком `log` подтверждённых операций. Реализуйте контекстный менеджер `transaction(db)` через `@contextmanager`: он должен собирать операции во временный список; при успешном завершении — добавлять их в `db.log`; при исключении — откатиться (отбросить список) и перебросить исключение.

from contextlib import contextmanager

class FakeDB:
    def __init__(self):
        self.log = []

@contextmanager
def transaction(db):
    pass


db = FakeDB()

with transaction(db) as tx:
    tx.append('INSERT user')
    tx.append('UPDATE balance')

print(db.log)  # ['INSERT user', 'UPDATE balance']

try:
    with transaction(db) as tx:
        tx.append('DELETE everything')
        raise RuntimeError('oops')
except RuntimeError:
    pass

print(db.log)  # всё ещё ['INSERT user', 'UPDATE balance'] — откат
Решение
from contextlib import contextmanager

class FakeDB:
    def __init__(self):
        self.log = []

@contextmanager
def transaction(db):
    pending = []
    try:
        yield pending
        db.log.extend(pending)  # подтверждение при успехе
    except Exception:
        pass  # откат — отбрасываем pending
        raise


db = FakeDB()

with transaction(db) as tx:
    tx.append('INSERT user')
    tx.append('UPDATE balance')

print(db.log)  # ['INSERT user', 'UPDATE balance']

try:
    with transaction(db) as tx:
        tx.append('DELETE everything')
        raise RuntimeError('oops')
except RuntimeError:
    pass

print(db.log)  # ['INSERT user', 'UPDATE balance']
08

ExitStack для динамических контекстных менеджеров

#

Используйте `contextlib.ExitStack`, чтобы открыть динамическое количество файлов одновременно и прочитать первую строку каждого. Имея список путей к файлам, откройте все внутри одного блока `with ExitStack()` и соберите первую строку каждого в список.

import contextlib

def read_first_lines(paths):
    pass


import tempfile, os

# Создаём тестовые файлы
paths = []
for i in range(3):
    p = tempfile.mktemp()
    with open(p, 'w') as f:
        f.write(f'строка из файла {i}\nещё строки')
    paths.append(p)

print(read_first_lines(paths))
# ['строка из файла 0', 'строка из файла 1', 'строка из файла 2']

for p in paths:
    os.remove(p)
Решение
import contextlib

def read_first_lines(paths):
    with contextlib.ExitStack() as stack:
        files = [stack.enter_context(open(p)) for p in paths]
        return [f.readline().strip() for f in files]


import tempfile, os

paths = []
for i in range(3):
    p = tempfile.mktemp()
    with open(p, 'w') as f:
        f.write(f'строка из файла {i}\nещё строки')
    paths.append(p)

print(read_first_lines(paths))

for p in paths:
    os.remove(p)
09

Контекстный менеджер с подавлением исключений

#

Напишите класс `Attempt`, который подавляет исключения `ValueError` внутри блока `with` и сохраняет исключение в `self.error`. Если исключение не произошло, `self.error` должен быть `None`. Другие типы исключений должны распространяться нормально.

class Attempt:
    def __init__(self):
        self.error = None

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        pass


with Attempt() as a:
    int('не число')

print(a.error)  # invalid literal for int() with base 10: 'не число'

with Attempt() as a:
    x = int('42')

print(a.error)  # None
Решение
class Attempt:
    def __init__(self):
        self.error = None

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is ValueError:
            self.error = exc_val
            return True  # подавляем исключение
        return False     # другие исключения распространяются


with Attempt() as a:
    int('не число')

print(a.error)

with Attempt() as a:
    x = int('42')

print(a.error)  # None
10

Потокобезопасный счётчик с Lock

#

Напишите класс `SafeCounter` с внутренним `threading.Lock`. Реализуйте метод `increment()`, использующий `with self.lock:` для безопасного увеличения `self.count`. Запустите 5 потоков, каждый из которых вызывает `increment()` 1000 раз, и убедитесь, что финальный счётчик равен 5000.

import threading

class SafeCounter:
    def __init__(self):
        self.count = 0
        self.lock = threading.Lock()

    def increment(self):
        pass


counter = SafeCounter()
threads = [threading.Thread(target=lambda: [counter.increment() for _ in range(1000)])
           for _ in range(5)]
for t in threads: t.start()
for t in threads: t.join()
print(counter.count)  # 5000
Решение
import threading

class SafeCounter:
    def __init__(self):
        self.count = 0
        self.lock = threading.Lock()

    def increment(self):
        with self.lock:
            self.count += 1


counter = SafeCounter()
threads = [threading.Thread(target=lambda: [counter.increment() for _ in range(1000)])
           for _ in range(5)]
for t in threads: t.start()
for t in threads: t.join()
print(counter.count)  # 5000