Динамическое добавление поведения через декораторы функций и классов. functools.wraps, contextlib.contextmanager, декораторы с параметрами.
Декораторы — это не магия, а просто функции, возвращающие функции.
Decorator — структурный паттерн для динамического добавления поведения объектам через обёртку. В Python также языковая конструкция для модификации функций.
# ❌ ПЛОХО: Взрыв классов
class Window:
def draw(self):
print("Drawing window")
class ScrollableWindow(Window):
def draw(self):
super().draw()
print("Adding scrollbar")
class BorderedWindow(Window):
def draw(self):
super().draw()
print("Adding border")
# А если нужно и scrollbar, и border?
class BorderedScrollableWindow(Window):
def draw(self):
super().draw()
print("Adding border")
print("Adding scrollbar")
# 2 фичи = 4 класса, 3 фичи = 8 классов...# ✅ ХОРОШО: Композиция обёрток
class Window:
def draw(self):
print("Drawing window")
class WindowDecorator:
def __init__(self, window: Window):
self._window = window
def draw(self):
self._window.draw()
class ScrollableWindow(WindowDecorator):
def draw(self):
super().draw()
print("Adding scrollbar")
class BorderedWindow(WindowDecorator):
def draw(self):
super().draw()
print("Adding border")
# Использование — комбинируем в runtime
window = Window()
window = ScrollableWindow(window)
window = BorderedWindow(window)
window.draw()
# Drawing window
# Adding scrollbar
# Adding borderfrom functools import wraps
def log_decorator(func):
@wraps(func) # Сохраняет имя и документацию оригинала
def wrapper(*args, **kwargs):
print(f"Calling {func.__name__}")
result = func(*args, **kwargs)
print(f"{func.__name__} returned {result}")
return result
return wrapper
@log_decorator
def add(a, b):
"""Add two numbers"""
return a + b
add(2, 3)
# Calling add
# add returned 5Важно: @wraps(func) сохраняет __name__, __doc__ и другие атрибуты оригинальной функции.
from functools import wraps
def retry(max_attempts: int = 3):
"""Декоратор с параметрами"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except Exception as e:
last_exception = e
print(f"Attempt {attempt + 1} failed: {e}")
raise last_exception
return wrapper
return decorator
@retry(max_attempts=5)
def flaky_operation():
import random
if random.random() < 0.7:
raise ConnectionError("Network error")
return "Success!"from functools import wraps
import time
def timing(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
end = time.perf_counter()
print(f"{func.__name__} took {end - start:.4f} seconds")
return result
return wrapper
@timing
def slow_function():
time.sleep(1)
return "Done"
slow_function() # slow_function took 1.0001 secondsclass Timer:
def __enter__(self):
import time
self.start = time.perf_counter()
return self
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 # Не подавляем исключения
with Timer() as t:
time.sleep(1)
# Elapsed: 1.0001sfrom contextlib import contextmanager
import time
@contextmanager
def timer():
"""Генератор как контекстный менеджер"""
start = time.perf_counter()
try:
yield # Возвращаем управление в with
finally:
elapsed = time.perf_counter() - start
print(f"Elapsed: {elapsed:.4f}s")
with timer():
time.sleep(1)
# Elapsed: 1.0001sВажно: Код до yield выполняется при входе в with, после yield — при выходе (даже при исключении).
from contextlib import contextmanager
import sqlite3
@contextmanager
def get_connection(db_path: str):
"""Контекстный менеджер для БД"""
conn = sqlite3.connect(db_path)
try:
yield conn # Возвращаем соединение в with
finally:
conn.close()
with get_connection("mydb.sqlite") as conn:
cursor = conn.execute("SELECT * FROM users")
users = cursor.fetchall()
# conn автоматически закроетсяfrom contextlib import contextmanager
@contextmanager
def transaction(db):
"""Транзакция савтоматическим rollback"""
try:
yield db
db.commit()
print("Transaction committed")
except Exception:
db.rollback()
print("Transaction rolled back")
raise
# Использование
with transaction(db) as db:
db.execute("INSERT INTO users VALUES (1, 'Alice')")
db.execute("UPDATE accounts SET balance = 100 WHERE user_id = 1")
# commit() вызывается автоматически
# rollback() при исключенииfrom functools import wraps
from typing import Callable, Any
def cache(func: Callable) -> Callable:
"""Простой кэш для функций"""
_cache = {}
@wraps(func)
def wrapper(*args) -> Any:
if args not in _cache:
_cache[args] = func(*args)
return _cache[args]
return wrapper
@cache
def fibonacci(n: int) -> int:
if n < 2:
return n
return fibonacci(n - 1) + fibonacci(n - 2)
print(fibonacci(100)) # Мгновенно после первого вызоваfrom functools import wraps
from typing import Optional
_current_user: Optional[str] = None
def login(user: str):
global _current_user
_current_user = user
def logout():
global _current_user
_current_user = None
def require_auth(func):
@wraps(func)
def wrapper(*args, **kwargs):
if _current_user is None:
raise PermissionError("Authentication required")
return func(*args, **kwargs)
return wrapper
@require_auth
def delete_account():
print(f"Deleting account for {_current_user}")
login("alice")
delete_account() # Deleting account for alice
logout()
delete_account() # PermissionError!from functools import wraps
import logging
logging.basicConfig(level=logging.ERROR)
def log_exceptions(func):
@wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
logging.error(f"Exception in {func.__name__}: {e}")
raise
return wrapper
@log_exceptions
def risky_operation():
raise ValueError("Something went wrong")from contextlib import contextmanager
import os
@contextmanager
def change_directory(path: str):
"""Временно меняет рабочую директорию"""
original = os.getcwd()
try:
os.chdir(path)
yield
finally:
os.chdir(original)
with change_directory("/tmp"):
print(os.getcwd()) # /tmp
print(os.getcwd()) # Вернулись в оригинальнуюfrom contextlib import contextmanager
@contextmanager
def suppress(*exceptions):
"""Подавляет указанные исключения"""
try:
yield
except exceptions:
pass # Игнорируем
with suppress(FileNotFoundError):
os.remove("nonexistent.txt") # Не вызывает ошибкуfrom functools import wraps
def decorator1(func):
@wraps(func)
def wrapper(*args, **kwargs):
print("Decorator 1: before")
result = func(*args, **kwargs)
print("Decorator 1: after")
return result
return wrapper
def decorator2(func):
@wraps(func)
def wrapper(*args, **kwargs):
print("Decorator 2: before")
result = func(*args, **kwargs)
print("Decorator 2: after")
return result
return wrapper
@decorator1
@decorator2
def greet():
print("Hello!")
greet()
# Decorator 1: before
# Decorator 2: before
# Hello!
# Decorator 2: after
# Decorator 1: afterПорядок: Декораторы применяются снизу вверх (@decorator2, затем @decorator1). Выполнение идёт в обратном порядке.
from contextlib import closing
import urllib.request
with closing(urllib.request.urlopen("http://example.com")) as page:
content = page.read()
# urlopen автоматически закроетсяfrom contextlib import ExitStack
with ExitStack() as stack:
file1 = stack.enter_context(open("file1.txt"))
file2 = stack.enter_context(open("file2.txt"))
file3 = stack.enter_context(open("file3.txt"))
# Все файлы закроются автоматическиfrom contextlib import redirect_stdout
from io import StringIO
f = StringIO()
with redirect_stdout(f):
print("Hello")
output = f.getvalue()
print(output) # Hello\n| Паттерн | Когда использовать | Pythonic-реализация |
|---|---|---|
| Decorator (класс) | Динамическое добавление поведения объектам | Композиция + общий интерфейс |
| @decorator (функция) | Модификация функций/методов | @wraps + wrapper |
| @contextmanager | Контекстные менеджеры без класса | Генератор с yield |
| ExitStack | Динамическое управление контекстами | contextlib.ExitStack |
Главный принцип: Декораторы добавляют поведение, не изменяя оригинальный код.
Изучите тему Flyweight и slots для оптимизации памяти.
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.