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