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