Python · Синтаксис · Продвинутый уровень
Контекстные менеджеры
Объекты, управляющие настройкой и очисткой через оператор `with`. Охватывает `__enter__`/`__exit__`, `contextlib.contextmanager` и паттерны управления ресурсами.
Краткое введение в тему и пояснения перед упражнениями (упражнения ниже):
@contextmanager, suppress, ExitStack и вложение
#класс vs @contextmanager, реальные паттерны, nullcontext, ошибки
#Упражнения:
Контекстный менеджер 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')
Управляемый менеджер файлов
#Напишите контекстный менеджер `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())
Декоратор @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")
Контекстный менеджер 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
Контекстный менеджер временного файла
#Напишите контекстный менеджер `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
Контекстный менеджер с отступом вывода
#Напишите контекстный менеджер `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('возврат на верхний уровень')
Контекстный менеджер транзакций
#Напишите класс `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']
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)
Контекстный менеджер с подавлением исключений
#Напишите класс `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
Потокобезопасный счётчик с 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