Замыкания и декораторы

Замыкания и декораторы в Python

Маршрут: Инженерная основа · тема 4 из 8

Навыки: M3, M6

До урока: functions, python_testing.

Результат: декораторы, сохраняющие поведение, служебные сведения и сигнатуру исходного вызываемого объекта.

Основа: замыкания, позднее связывание и wraps. Работа в эксплуатации: настраиваемые декораторы, правила сбоев и тестируемость. Углубление: ячейки замыканий, дескрипторы и типизация через ParamSpec.

Подтверждение: декоратор с сохранённой сигнатурой и тестами сбоев.

Замыкание сохраняет ссылки на имена из внешней функции. Декоратор принимает вызываемый объект и возвращает замену, которая добавляет поведение или меняет регистрацию исходного объекта.

#1. Замыкания (Closures)

Замыкание возникает когда внутренняя функция ссылается на переменные из внешней области видимости, и эта внешняя функция уже завершилась.

def make_counter(start=0):
    count = start   # переменная в замыкании

    def increment():
        nonlocal count   # без nonlocal — count не изменится (только чтение)
        count += 1
        return count

    def reset():
        nonlocal count
        count = start

    return increment, reset

inc, rst = make_counter(10)
print(inc())   # 11
print(inc())   # 12
print(inc())   # 13
rst()
print(inc())   # 11 — сброшен

#Инспекция замыкания

def make_adder(x):
    def adder(y):
        return x + y
    return adder

add5 = make_adder(5)

# Просмотр захваченных переменных
print(add5.__closure__)           # (<cell at 0x...>,)
print(add5.__closure__[0].cell_contents)  # 5

# Свободные переменные
print(add5.__code__.co_freevars)  # ('x',)

#`nonlocal` — изменение переменной из внешней области

def make_accumulator():
    total = 0

    def add(n):
        nonlocal total
        total += n
        return total

    return add

acc = make_accumulator()
print(acc(10))   # 10
print(acc(5))    # 15
print(acc(20))   # 35

# Без nonlocal: UnboundLocalError (при попытке total += n)
# Потому что total += n эквивалентно total = total + n
# Python видит total = ... и считает total локальной переменной

#Ловушка: позднее связывание в цикле

# Проблема: Проблема: все лямбды замыкаются на ОДНУ переменную i
funcs = [lambda: i for i in range(5)]
print([f() for f in funcs])   # [4, 4, 4, 4, 4] — не [0,1,2,3,4]!

# Почему: i — одна переменная, при вызове f() она уже = 4

# Вариант: Решение 1: параметр по умолчанию (захват значения, не ссылки)
funcs = [lambda i=i: i for i in range(5)]
print([f() for f in funcs])   # [0, 1, 2, 3, 4]

# Вариант: Решение 2: фабричная функция
def make_func(value):
    return lambda: value

funcs = [make_func(i) for i in range(5)]
print([f() for f in funcs])   # [0, 1, 2, 3, 4]

# Та же ловушка с closures:
callbacks = []
for i in range(3):
    callbacks.append(lambda: print(i))   # все напечатают 2!

# Правильно:
callbacks = [lambda i=i: print(i) for i in range(3)]

#Практические применения замыканий

# 1. Фабрики функций
def make_validator(min_val, max_val):
    def validate(value):
        if not min_val <= value <= max_val:
            raise ValueError(f"Value {value} not in [{min_val}, {max_val}]")
        return value
    return validate

validate_age = make_validator(0, 150)
validate_score = make_validator(0, 100)

# 2. Частичное применение функций (manual partial)
def multiply(x, y):
    return x * y

double = lambda y: multiply(2, y)
triple = lambda y: multiply(3, y)

# Лучше через functools.partial:
from functools import partial
double = partial(multiply, 2)

# 3. Состояние без классов (иногда)
def make_rate_limiter(max_calls_per_second):
    import time
    last_calls = []

    def limit():
        now = time.monotonic()
        # Оставляем только вызовы за последнюю секунду
        recent = [t for t in last_calls if now - t < 1.0]
        if len(recent) >= max_calls_per_second:
            raise RuntimeError("Rate limit exceeded")
        last_calls.clear()
        last_calls.extend(recent)
        last_calls.append(now)

    return limit

#2. Декораторы — базовый паттерн

Декоратор — функция, принимающая функцию и возвращающая функцию.

@decorator        # синтаксический сахар для:
def func(): ...   # func = decorator(func)

import time
from functools import wraps

def timer(func):
    @wraps(func)   # сохраняет __name__, __doc__, __module__ оригинала
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"{func.__name__} took {elapsed:.4f}s")
        return result
    return wrapper

@timer
def slow_function(n):
    """Имитирует медленную работу."""
    time.sleep(n)
    return n * 2

result = slow_function(0.1)
print(result)                        # 0.2
print(slow_function.__name__)        # "slow_function" (не "wrapper"!)
print(slow_function.__doc__)         # "Имитирует медленную работу."

# Без @wraps:
# slow_function.__name__ == "wrapper" — теряем метаданные
# Важно для отладки, логов, pytest, sphinx

#3. Декораторы с аргументами

Нужен дополнительный уровень вложенности:

from functools import wraps

# Паттерн: decorator factory
def retry(max_attempts=3, delay=1.0, exceptions=(Exception,)):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exc = None
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exc = e
                    if attempt < max_attempts - 1:
                        print(f"Attempt {attempt + 1} failed: {e}. Retrying...")
                        time.sleep(delay * (2 ** attempt))   # exponential backoff
            raise last_exc
        return wrapper
    return decorator

@retry(max_attempts=3, delay=0.5, exceptions=(ConnectionError, TimeoutError))
def fetch_data(url):
    return requests.get(url).json()

# Декоратор поддерживающий оба варианта вызова: @timer и @timer(label="...")
def timer(func=None, *, label=None):
    def decorator(f):
        @wraps(f)
        def wrapper(*args, **kwargs):
            lbl = label or f.__name__
            start = time.perf_counter()
            result = f(*args, **kwargs)
            print(f"{lbl}: {time.perf_counter() - start:.4f}s")
            return result
        return wrapper

    if func is not None:
        return decorator(func)   # @timer без аргументов
    return decorator             # @timer(label="...") с аргументами

@timer                          # OK
def func1(): pass

@timer(label="custom")         # OK
def func2(): pass

#4. Стекирование декораторов

@decorator1
@decorator2
@decorator3
def func():
    pass

# Эквивалентно:
func = decorator1(decorator2(decorator3(func)))
# Порядок применения: decorator3 → decorator2 → decorator1

# Практический пример:
@timer
@retry(max_attempts=3)
@validate_input
def process_data(data):
    return transform(data)

# При вызове process_data(x):
# 1. validate_input оборачивает оригинал
# 2. retry оборачивает validated
# 3. timer оборачивает retried
# → timer → retry → validate_input → оригинал

#5. Классовые декораторы

Используйте класс с __call__ когда нужно хранить состояние:

class CountCalls:
    def __init__(self, func):
        wraps(func)(self)   # копируем метаданные
        self.func = func
        self.call_count = 0

    def __call__(self, *args, **kwargs):
        self.call_count += 1
        return self.func(*args, **kwargs)

    def reset(self):
        self.call_count = 0

@CountCalls
def expensive_operation(n):
    return n ** 2

expensive_operation(5)
expensive_operation(10)
print(expensive_operation.call_count)   # 2
expensive_operation.reset()
print(expensive_operation.call_count)   # 0

# Классовый декоратор с аргументами:
class Rate:
    def __init__(self, calls_per_second):
        self.calls_per_second = calls_per_second
        self._calls = []

    def __call__(self, func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            import time
            now = time.monotonic()
            self._calls = [t for t in self._calls if now - t < 1.0]
            if len(self._calls) >= self.calls_per_second:
                raise RuntimeError("Rate limit exceeded")
            self._calls.append(now)
            return func(*args, **kwargs)
        return wrapper

@Rate(10)  # 10 вызовов в секунду
def api_call():
    pass

#6. Декораторы классов

Декораторы можно применять и к классам:

def add_repr(cls):
    """Добавляет __repr__ показывающий все атрибуты."""
    def __repr__(self):
        attrs = ", ".join(f"{k}={v!r}" for k, v in vars(self).items())
        return f"{cls.__name__}({attrs})"
    cls.__repr__ = __repr__
    return cls

@add_repr
class Config:
    def __init__(self, host, port):
        self.host = host
        self.port = port

print(Config("localhost", 8080))   # Config(host='localhost', port=8080)

# Полезный декоратор: singleton
def singleton(cls):
    instances = {}
    @wraps(cls)
    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    return get_instance

@singleton
class AppConfig:
    def __init__(self):
        self.settings = {}

c1 = AppConfig()
c2 = AppConfig()
assert c1 is c2   # True

#7. `functools` — полезные декораторы

#`lru_cache` и `cache`

from functools import lru_cache, cache

# lru_cache с ограниченным размером кэша
@lru_cache(maxsize=128)
def fibonacci(n: int) -> int:
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

print(fibonacci(50))          # Мгновенно
print(fibonacci.cache_info()) # CacheInfo(hits=48, misses=51, maxsize=128, currsize=51)
fibonacci.cache_clear()       # Очистить кэш

# cache (Python 3.9+) — без ограничений, немного быстрее
@cache
def factorial(n: int) -> int:
    return n * factorial(n - 1) if n else 1

#`singledispatch` — перегрузка по типу

from functools import singledispatch

@singledispatch
def process(value):
    raise NotImplementedError(f"No handler for {type(value)}")

@process.register(int)
def _(value: int):
    return f"Integer: {value * 2}"

@process.register(str)
def _(value: str):
    return f"String: {value.upper()}"

@process.register(list)
def _(value: list):
    return f"List of {len(value)} items"

print(process(42))         # "Integer: 84"
print(process("hello"))    # "String: HELLO"
print(process([1, 2, 3]))  # "List of 3 items"

#`partial` — частичное применение

from functools import partial

def power(base, exp):
    return base ** exp

square = partial(power, exp=2)
cube = partial(power, exp=3)

print(square(5))   # 25
print(cube(3))     # 27

# Применение для методов класса
class Logger:
    def log(self, level, message):
        print(f"[{level}] {message}")

logger = Logger()
info = partial(logger.log, "INFO")
error = partial(logger.log, "ERROR")
info("Server started")   # [INFO] Server started

#8. Декораторы в реальных проектах

#Flask / FastAPI routes

from flask import Flask
app = Flask(__name__)

@app.route("/users/<int:id>", methods=["GET"])
def get_user(id: int):
    return {"id": id}

# app.route — это декоратор фабрика:
# def route(path, methods=None):
#     def decorator(func):
#         # регистрирует func как обработчик пути
#         return func
#     return decorator

#Аутентификация

from functools import wraps
from flask import request, jsonify

def require_auth(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        token = request.headers.get("Authorization", "").replace("Bearer ", "")
        if not verify_token(token):
            return jsonify({"error": "Unauthorized"}), 401
        return f(*args, **kwargs)
    return decorated

@app.route("/admin")
@require_auth
def admin_panel():
    return {"message": "Welcome, admin!"}

#Кэширование с TTL

import time
from functools import wraps

def cached(ttl_seconds=60):
    """Кэширует результат функции на ttl_seconds секунд."""
    cache = {}

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            key = (args, tuple(sorted(kwargs.items())))
            if key in cache:
                result, timestamp = cache[key]
                if time.monotonic() - timestamp < ttl_seconds:
                    return result
            result = func(*args, **kwargs)
            cache[key] = (result, time.monotonic())
            return result

        wrapper.cache_clear = lambda: cache.clear()
        return wrapper
    return decorator

@cached(ttl_seconds=300)   # Кэш на 5 минут
def get_user_from_db(user_id: int):
    # Синтаксис placeholder зависит от драйвера; значение не склеивается с SQL.
    return db.execute(
        "SELECT * FROM users WHERE id = %s",
        (user_id,),
    ).fetchone()

#Логирование

import logging
from functools import wraps

def log_calls(logger=None, level="DEBUG"):
    if logger is None:
        logger = logging.getLogger(__name__)
    log_func = getattr(logger, level.lower())

    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            log_func(f"Calling {func.__name__}({args!r}, {kwargs!r})")
            try:
                result = func(*args, **kwargs)
                log_func(f"{func.__name__} returned {result!r}")
                return result
            except Exception as e:
                logger.exception(f"{func.__name__} raised {e!r}")
                raise
        return wrapper
    return decorator

@log_calls(level="INFO")
def divide(a, b):
    return a / b

#9. Lambda-функции

# Lambda: анонимная функция, одно выражение
double = lambda x: x * 2
print(double(5))   # 10

# Применения: key функции
data = [{"name": "Charlie", "age": 30}, {"name": "Alice", "age": 25}]
sorted_by_name = sorted(data, key=lambda d: d["name"])
sorted_by_age = sorted(data, key=lambda d: d["age"], reverse=True)

# map / filter (предпочитайте list comprehension в большинстве случаев)
numbers = [1, 2, 3, 4, 5]
doubled = list(map(lambda x: x * 2, numbers))
evens = list(filter(lambda x: x % 2 == 0, numbers))

# То же через comprehensions (более Pythonic):
doubled = [x * 2 for x in numbers]
evens = [x for x in numbers if x % 2 == 0]

# functools.reduce с lambda
from functools import reduce
product = reduce(lambda acc, x: acc * x, numbers, 1)   # 1*2*3*4*5 = 120

#Когда НЕ использовать lambda

# Проблема: Присваивание lambda переменной — лучше def
process = lambda x, y: x + y   # PEP 8 против этого

# Вариант: Используйте def
def process(x, y):
    return x + y

# Проблема: Сложная логика в lambda
transform = lambda x: (x * 2 if x > 0 else x / 2) + len(str(x))

# Вариант: Дайте ей имя и место
def transform(x: int | float) -> float:
    base = x * 2 if x > 0 else x / 2
    return base + len(str(x))

#10. `functools.wraps` — под капотом

from functools import wraps, WRAPPER_ASSIGNMENTS, WRAPPER_UPDATES

# update_wrapper, который вызывает wraps, присваивает перечисленные атрибуты:
print(WRAPPER_ASSIGNMENTS)
# Точный кортеж зависит от версии: в Python 3.14 ленивые аннотации представлены
# через __annotate__, а в более ранних версиях копировался __annotations__.
# __type_params__ включён начиная с Python 3.12.

# А эти отображения обновляет, а не присваивает:
WRAPPER_UPDATES      # ('__dict__',)

# __wrapped__ устанавливается отдельно и указывает на исходную функцию.

def my_decorator(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

@my_decorator
def foo():
    """Docstring."""
    pass

# Доступ к оригинальной функции:
print(foo.__wrapped__)    # <function foo at 0x...> — оригинал!
print(foo.__wrapped__())  # вызов без декоратора

# Полезно для тестирования:
def test_foo_directly():
    result = foo.__wrapped__()   # обойти декоратор в тестах

#Практика: от пробы к рабочему решению

#Шаг 1. Соберите минимальный пример

Замыкание должно захватывать именно то состояние, которое заявлено контрактом, а декоратор — сохранять сведения об исходной функции. Результат основы — обёртка без ошибки позднего связывания.

#Шаг 2. Проверьте поведение

import inspect
from functools import wraps


def traced(function):
    @wraps(function)
    def wrapper(*args, **kwargs):
        return function(*args, **kwargs)

    return wrapper


callbacks = [lambda value=index: value for index in range(3)]
assert [callback() for callback in callbacks] == [0, 1, 2]

@traced
def add(left: int, right: int = 0) -> int:
    return left + right


assert add.__name__ == "add"
assert inspect.signature(add) == inspect.signature(add.__wrapped__)

#Шаг 3. Доведите решение до рабочего сценария

Практика с подсказками: напишите настраиваемый декоратор повторов только для перечисленных временных ошибок. Внедрите функцию ожидания, сохраните metadata и проверьте нулевое число повторов, успешный повтор и неожиданный сбой.

#Шаг 4. Объясните внутренний механизм

Опишите тип обёртки через ParamSpec и исследуйте closure cells. Сравните декоратор функции с дескриптором метода: где появляется связывание self и какие свойства теряются при наивной обёртке.

#Упражнения

Напишите декоратор @validate_types который проверяет что аргументы функции соответствуют аннотированным типам (используйте func.__annotations__).
Реализуйте @memoize без functools.lru_cache — с помощью словаря в замыкании. Обеспечьте работу с именованными аргументами.
Создайте декоратор @timeout(seconds) для синхронных функций используя threading.Timer или сигналы.
Напишите @deprecated(message, version) — декоратор выдающий DeprecationWarning при вызове функции.
Объясните: почему @wraps(func) важен для pytest и sphinx? Что сломается без него?
Реализуйте @once — функция выполняется только при первом вызове, последующие вызовы возвращают кэшированный результат.

#Самопроверка

validate_types должен явно описать ограничения: простая проверка через isinstance не реализует весь typing (list[int], объединения, Annotated). Добавьте тест неподдерживаемой аннотации.
Ключ memoize обязан различать позиционные и именованные вызовы с одинаковым смыслом либо документировать, что они дают разные записи; нехэшируемый аргумент обрабатывается явно.
Сигнальный timeout работает только в главном потоке Unix; вариант с threading.Timer не может принудительно остановить произвольную функцию. Зафиксируйте выбранное ограничение в контракте.
Все декораторы сохраняют __name__, __doc__, __wrapped__ и исходную сигнатуру.
Для once определите поведение после первой ошибки и при одновременных вызовах из нескольких потоков.

#Практическая лаборатория

Закрепите тему в лаборатории «Наблюдающий декоратор без изменения семантики». Вы сохраните сигнатуру через ParamSpec, не превратите обычную функцию в корутину и после темы асинхронности отдельно докажете корректную отмену.

git clone --branch v0.6.2 --depth 1 https://gitlab.potapov.me/courses/python-labs.git
cd python-labs
uv sync --group test --group quality
uv run --group test pytest closures_decorators/tests