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