Замыкания, декораторы функций, lambda
Замыкание — функция, помнящая своё окружение. Декоратор — синтаксический сахар для оборачивания функции. Оба механизма лежат в основе Flask, FastAPI, pytest, click.
Замыкание возникает когда внутренняя функция ссылается на переменные из внешней области видимости, и эта внешняя функция уже завершилась.
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Декоратор — функция, принимающая функцию и возвращающая функцию.
@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Нужен дополнительный уровень вложенности:
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@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 → оригиналИспользуйте класс с __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Декораторы можно применять и к классам:
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 # Truefunctools — полезные декораторыlru_cache и cachefrom 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 1singledispatch — перегрузка по типу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 startedfrom 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 decoratorfrom 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!"}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):
return db.query(f"SELECT * FROM users WHERE id = {user_id}")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# 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 переменной — лучше 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))functools.wraps — под капотомfrom functools import wraps, WRAPPER_ASSIGNMENTS
# wraps копирует эти атрибуты:
WRAPPER_ASSIGNMENTS # ('__module__', '__name__', '__qualname__', '__annotations__',
# '__doc__', '__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__() # обойти декоратор в тестах@validate_types который проверяет что аргументы функции соответствуют аннотированным типам (используйте func.__annotations__).@memoize без functools.lru_cache — с помощью словаря в замыкании. Обеспечьте работу с именованными аргументами.@timeout(seconds) для синхронных функций используя threading.Timer или сигналы.@deprecated(message, version) — декоратор выдающий DeprecationWarning при вызове функции.@wraps(func) важен для pytest и sphinx? Что сломается без него?@once — функция выполняется только при первом вызове, последующие вызовы возвращают кэшированный результат.Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.