Года три назад отлаживал систему обработки платежей. Нужно было логировать каждую транзакцию с уникальным префиксом - ID сессии, который формировался при создании обработчика. Классический вариант с классом казался избыточным для такой мелочи. И тут я вспомнил про замыкания - штука, которую изучал когда-то теоретически, но толком не применял.
Замыкание - это функция, которая "запоминает" переменные из области видимости, в которой была создана, даже после того как эта область перестала существовать. Звучит запутанно? На практике всё проще.
| Python | 1
2
3
4
5
6
7
8
9
10
11
| def create_logger(session_id):
# session_id захватывается замыканием
def log(message):
print(f"[Session {session_id}] {message}")
return log
logger1 = create_logger("ABC123")
logger2 = create_logger("XYZ789")
logger1("Payment initiated") # [Session ABC123] Payment initiated
logger2("Payment initiated") # [Session XYZ789] Payment initiated |
|
create_logger, она возвращает внутреннюю функцию log. Но эта log не просто болтается в памяти сама по себе - она тащит с собой значение session_id. Причем для каждого вызова create_logger создается свой независимый экземпляр с собственной копией переменной.
Python реализует это через атрибут __closure__ у функции-объекта. Там хранятся ячейки памяти с захваченными значениями. Можно даже посмотреть:
| Python | 1
2
| print(logger1.__closure__) # (<cell at 0x...: str object at 0x...>,)
print(logger1.__closure__[0].cell_contents) # ABC123 |
|
Механика замыканий: что происходит под капотом
Когда Python встречает вложенную функцию, он не просто создаёт объект и забывает. Происходит анализ: какие переменные из внешней области понадобятся внутренней функции? Для каждой такой переменной создается специальная структура - ячейка (cell), которая становится мостом между областями видимости. Проверял это на днях, отлаживая систему кэширования. Функция должна была запоминать последний результат, но при определенных условиях сбрасывать кэш. Начал с простого варианта и застрял на понимании, почему изменение счетчика иногда падает с UnboundLocalError.
Область видимости и захват переменных
Python определяет область видимости статически, на этапе компиляции байт-кода. Когда интерпретатор парсит функцию, он строит таблицу имен - что локальное, что нужно искать во внешних областях. Это важно: решение принимается ДО выполнения кода.
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| def outer():
x = 10
y = 20
def inner():
# x захватывается - используется, но не изменяется
print(x)
# z локальная - создается внутри
z = 30
return x + z
# Смотрим, что захвачено
print(inner.__code__.co_freevars) # ('x',)
print(inner.__code__.co_varnames) # ('z',)
return inner
fn = outer()
print(fn.__closure__) # (<cell at 0x...: int object at 0x...>,)
print(fn.__closure__[0].cell_contents) # 10 |
|
co_freevars содержит имена переменных, которые функция берет из внешней области. co_varnames - локальные переменные. Переменная y вообще не попала в список - она не используется во inner, поэтому не захватывается. Python оптимизирует и не таскает лишнее.
Но есть подвох. Попробуйте изменить x внутри inner:
| Python | 1
2
3
4
5
6
7
8
| def outer():
x = 10
def inner():
x = x + 1 # UnboundLocalError: local variable 'x' referenced before assignment
return x
return inner |
|
x = ... и решает: "x тут локальная". Но в правой части выражения x + 1 он пытается прочитать эту локальную переменную, которая еще не инициализирована. Парадокс? Нет - последствие статического анализа.
Правило LEGB и связывание имен
Когда интерпретатор встречает имя, он ищет его по правилу LEGB: Local → Enclosing → Global → Built-in. Это не просто порядок поиска - это четко определенные области с разными правилами доступа.
Local - локальная область функции. Здесь живут параметры и локальные переменные.
Enclosing - замыкающая область. Все функции-родители текущей функции. Именно отсюда берутся захваченные переменные.
Global - модульная область видимости. Переменные, объявленные на уровне модуля.
Built-in - встроенные имена Python (len, print, ValueError и т.д.).
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
| builtin_name = len # Built-in
global_var = "глобальная"
def outer():
enclosing_var = "внешняя"
def middle():
middle_var = "средняя"
def inner():
local_var = "локальная"
# Поиск идет в таком порядке
print(local_var) # Local
print(middle_var) # Enclosing (ближайшая)
print(enclosing_var) # Enclosing (дальше)
print(global_var) # Global
print(builtin_name) # Built-in
return inner
return middle
fn = outer()()
fn() |
|
nonlocal. Без этого Python создаст новую локальную переменную с таким же именем.
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
| def counter():
count = 0
def increment():
nonlocal count # Явно говорим: хочу изменить внешнюю count
count += 1
return count
return increment
cnt = counter()
print(cnt()) # 1
print(cnt()) # 2
print(cnt()) # 3 |
|
nonlocal появилось в Python 3. До этого приходилось использовать трюки - например, упаковывать изменяемое значение в список или словарь. Изменяемые объекты можно модифицировать без nonlocal, потому что мы не переприсваиваем саму переменную:
| Python | 1
2
3
4
5
6
7
8
| def counter_old_style():
state = {'count': 0} # Изменяемый объект
def increment():
state['count'] += 1 # Изменяем содержимое, не переменную state
return state['count']
return increment |
|
Жизненный цикл и управление памятью
Замыкания влияют на сборку мусора. Обычно когда функция завершается, её локальные переменные уничтожаются. Но если внутренняя функция захватила эти переменные - они остаются в памяти, пока существует ссылка на замыкание.
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| import sys
def create_closure():
# Большой объект в enclosing области
large_data = [0] * 1_000_000
def inner():
return len(large_data) # Захватывает large_data
return inner
fn = create_closure()
# large_data все еще в памяти!
print(sys.getsizeof(fn.__closure__[0].cell_contents)) # ~8000000 байт
# Освобождаем память
del fn |
|
weakref).
Python использует подсчет ссылок для управления памятью. У каждого объекта есть счетчик - сколько ссылок на него существует. Когда счетчик обнуляется, объект уничтожается. Но замыкания создают дополнительные ссылки:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
| import sys
def outer():
data = [1, 2, 3]
print(f"Ссылок на data: {sys.getrefcount(data) - 1}") # 1 (минус временная в getrefcount)
def inner():
return data[0]
print(f"После создания inner: {sys.getrefcount(data) - 1}") # 2 (+ ссылка из замыкания)
return inner
fn = outer()
# data все еще жива через fn.__closure__ |
|
| Python | 1
2
3
4
5
6
7
8
9
10
| def create_circular():
container = {}
def inner():
return container
container['func'] = inner # Циклическая ссылка!
return inner
# container -> inner -> container через __closure__ |
|
Когда копался в байт-коде замыканий для оптимизации критичного участка системы, обнаружил интересную деталь. Python создает специальную инструкцию LOAD_DEREF для доступа к захваченным переменным. Это медленнее обычного LOAD_FAST для локальных переменных - примерно на 10-15%.
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
| import dis
def outer():
x = 42
def with_closure():
return x # LOAD_DEREF
def without_closure():
y = 42
return y # LOAD_FAST
return with_closure, without_closure
fn_closure, fn_local = outer()
print("С замыканием:")
dis.dis(fn_closure)
[H2]LOAD_DEREF 0 (x)[/H2]
print("\nБез замыкания:")
dis.dis(fn_local)
# LOAD_FAST 0 (y) |
|
__slots__ - прирост составил около 8% на миллионе итераций. Но за это платишь многословностью кода.
Замыкания и инструментарий отладки
Отладка замыканий бывает болью. Дебаггер показывает локальные переменные, но захваченные часто прячутся. В PyCharm можно развернуть __closure__ и посмотреть содержимое ячеек, но это не интуитивно.
Написал как-то вспомогательную функцию для инспекции:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
| def inspect_closure(func):
"""Показывает все захваченные переменные замыкания"""
if not func.__closure__:
print(f"{func.__name__} не является замыканием")
return
freevars = func.__code__.co_freevars
cells = func.__closure__
print(f"Замыкание {func.__name__} захватывает:")
for name, cell in zip(freevars, cells):
try:
value = cell.cell_contents
print(f" {name} = {value!r} (id: {id(value)})")
except ValueError:
print(f" {name} = <не инициализировано>")
def factory(a, b):
def worker(x):
return a * x + b
return worker
fn = factory(2, 5)
inspect_closure(fn)
# Замыкание worker захватывает:
# a = 2 (id: 140...)
# b = 5 (id: 140...) |
|
cell_contents выбросит ValueError. Проверяй это явно.
Замыкания в многопоточной среде
Тут начинается веселье. Захваченные переменные НЕ защищены GIL автоматически при изменении. Если несколько потоков вызывают замыкание, модифицирующее захваченное состояние, получите гонку данных.
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
| import threading
import time
def create_counter():
count = 0
def increment():
nonlocal count
# Некорректная реализация без блокировки!
temp = count
time.sleep(0.0001) # Симулируем задержку
count = temp + 1
return count
return increment
counter = create_counter()
def worker():
for _ in range(100):
counter()
threads = [threading.Thread(target=worker) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
# Ожидаем 1000, получаем меньше
print(inspect_closure(counter)) |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
| import threading
def create_thread_safe_counter():
count = 0
lock = threading.Lock()
def increment():
nonlocal count
with lock:
count += 1
return count
return increment
# Или через thread-local
def create_local_counter():
local = threading.local()
def increment():
if not hasattr(local, 'count'):
local.count = 0
local.count += 1
return local.count
return increment |
|
Оптимизация: когда замыкания не нужны
Python создает объект-замыкание даже если ничего не захватывается. Проверял это профилировщиком:
| Python | 1
2
3
4
5
6
7
| def outer():
def inner():
pass # Ничего не захватывает
return inner
fn = outer()
print(fn.__closure__) # None - не замыкание |
|
Когда замыкание используется один раз и сразу - lambda может быть эффективнее:
| Python | 1
2
3
4
5
6
7
8
9
10
11
| # Замыкание - создается объект функции
def make_multiplier(factor):
def multiply(x):
return x * factor
return multiply
# Lambda - тоже создает объект, но синтаксис компактнее
make_multiplier_lambda = lambda factor: lambda x: x * factor
# Но для одноразового использования - прямой вызов
result = (lambda factor: (lambda x: x * factor)(5))(3) # 15 |
|
Особые случаи и граничные условия
Замыкания могут захватывать не только переменные, но и параметры с значениями по умолчанию:
| Python | 1
2
3
4
5
6
7
8
9
10
| def outer(x=10):
def inner():
return x # Захватывает параметр x
return inner
fn1 = outer()
fn2 = outer(20)
print(fn1()) # 10
print(fn2()) # 20 |
|
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
| def create_funcs():
funcs = []
for i in range(3):
def func():
return i # Захватывает i
funcs.append(func)
return funcs
# Все функции вернут 2 - последнее значение i!
for fn in create_funcs():
print(fn()) # 2, 2, 2
# Исправление через параметр по умолчанию
def create_funcs_fixed():
funcs = []
for i in range(3):
def func(x=i): # i вычисляется сразу
return x
funcs.append(func)
return funcs
for fn in create_funcs_fixed():
print(fn()) # 0, 1, 2 |
|
С помощью замыкания напишите генератор очередного простого числа из списка
С помощью замыкания напишите генератор очередного простого числа из заданного списка.
lst = ; func... Используя замыкания функций, объявить внутреннюю функцию
Доброго дня! Не могу решить задачку.
На вход программы поступает две строки: первая с тегом,... Замыкания
Доброго времени суток. Имеется такая задача на замыкания:
Используя замыкания функций, объявите... Используя замыкания функций, определите вложенную функцию
Используя замыкания функций, определите вложенную функцию, которая бы увеличивала
значение...
Практические сценарии применения
В прошлом году делал API для обработки больших JSON-файлов. Парсинг одного файла занимал секунд десять - клиентам приходилось ждать при каждом запросе. Очевидное решение - кэширование. Но нужен был кэш, который запоминает не только результат, но и время последней проверки файла. Класс писать лень, хотелось чего-то легковесного.
Кэширование с автоматической инвалидацией
Замыкание позволяет упаковать кэш и логику проверки в одну функцию-фабрику:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
| import time
from pathlib import Path
def create_file_cache(filepath, ttl=60):
"""Создаёт кэшированный парсер с TTL"""
cache = {'data': None, 'mtime': 0, 'cached_at': 0}
path = Path(filepath)
def parse():
nonlocal cache
current_mtime = path.stat().st_mtime
now = time.time()
# Проверяем: файл изменился или кэш устарел
if (cache['data'] is None or
cache['mtime'] != current_mtime or
now - cache['cached_at'] > ttl):
print(f"Парсим файл {filepath}")
cache['data'] = path.read_text() # Реальный парсинг
cache['mtime'] = current_mtime
cache['cached_at'] = now
else:
print("Возвращаем из кэша")
return cache['data']
return parse
parser = create_file_cache('data.json', ttl=30)
result1 = parser() # Парсим файл
result2 = parser() # Возвращаем из кэша
time.sleep(31)
result3 = parser() # TTL истек, парсим снова |
|
cache живёт между вызовами. Каждый парсер получает собственное хранилище - можно создать десяток парсеров для разных файлов, они не конфликтуют. Это проще и компактнее класса, особенно когда кэширование нужно добавить в существующий код без рефакторинга.
Развил эту идею дальше - сделал декоратор для мемоизации с ограничением размера кэша:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
| from collections import OrderedDict
def memoize(maxsize=128):
"""Декоратор для кэширования с LRU вытеснением"""
def decorator(func):
cache = OrderedDict()
def wrapper(*args):
# Используем args как ключ (должны быть хешируемыми)
if args in cache:
# Перемещаем в конец - недавно использованный
cache.move_to_end(args)
return cache[args]
result = func(*args)
cache[args] = result
# Вытесняем старые записи
if len(cache) > maxsize:
cache.popitem(last=False) # Удаляем самый старый
return result
wrapper.cache_info = lambda: {
'size': len(cache),
'maxsize': maxsize,
'keys': list(cache.keys())
}
wrapper.cache_clear = lambda: cache.clear()
return wrapper
return decorator
@memoize(maxsize=3)
def fibonacci(n):
if n < 2:
return n
return fibonacci(n-1) + fibonacci(n-2)
print(fibonacci(10)) # Вычисляется
print(fibonacci(10)) # Из кэша
print(fibonacci.cache_info()) # {'size': 3, 'maxsize': 3, ...} |
|
cache хранит результаты, maxsize задаёт лимит. Метод cache_info добавлен как атрибут функции - удобно для мониторинга.
Декораторы с параметрами и внутренним состоянием
Классический декоратор без параметров - простая функция. Но если нужны параметры конфигурации, появляется дополнительный уровень вложенности. Замыкание захватывает эти параметры.
Писал систему логирования, где уровень детализации задавался для каждой функции отдельно. Декоратор должен был считать вызовы и выводить статистику:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
| import functools
import time
def monitored(level='INFO', sample_rate=1.0):
"""Декоратор с мониторингом вызовов"""
def decorator(func):
stats = {'calls': 0, 'errors': 0, 'total_time': 0}
@functools.wraps(func)
def wrapper(*args, **kwargs):
nonlocal stats
stats['calls'] += 1
# Пропускаем вызовы согласно sample_rate
import random
if random.random() > sample_rate:
return func(*args, **kwargs)
start = time.perf_counter()
try:
result = func(*args, **kwargs)
elapsed = time.perf_counter() - start
stats['total_time'] += elapsed
if level == 'DEBUG':
print(f"[{level}] {func.__name__}: {elapsed:.4f}s")
return result
except Exception as e:
stats['errors'] += 1
print(f"[ERROR] {func.__name__}: {e}")
raise
# Добавляем метод для просмотра статистики
wrapper.get_stats = lambda: stats.copy()
return wrapper
return decorator
@monitored(level='DEBUG', sample_rate=0.5)
def process_data(data):
time.sleep(0.1) # Симуляция работы
return len(data)
for i in range(10):
process_data([1, 2, 3])
print(process_data.get_stats())
# {'calls': 10, 'errors': 0, 'total_time': ~0.5} |
|
monitored принимает параметры конфигурации. decorator получает декорируемую функцию. wrapper захватывает и level, и sample_rate, и stats, и оригинальную func. Без замыканий пришлось бы городить класс с __call__.
Фабрики функций под конкретные задачи
Иногда нужно генерировать множество похожих функций с разными параметрами. Классический случай - валидаторы. Делал форму с десятком полей, каждое со своими ограничениями.
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
| def create_validator(field_name, min_len=0, max_len=float('inf'), pattern=None):
"""Создаёт функцию-валидатор для поля формы"""
import re
compiled_pattern = re.compile(pattern) if pattern else None
def validate(value):
errors = []
if not isinstance(value, str):
errors.append(f"{field_name}: должно быть строкой")
return errors
if len(value) < min_len:
errors.append(f"{field_name}: минимум {min_len} символов")
if len(value) > max_len:
errors.append(f"{field_name}: максимум {max_len} символов")
if compiled_pattern and not compiled_pattern.match(value):
errors.append(f"{field_name}: неверный формат")
return errors
return validate
# Создаем валидаторы
validators = {
'username': create_validator('Username', min_len=3, max_len=20,
pattern=r'^[a-zA-Z0-9_]+$'),
'email': create_validator('Email', min_len=5, max_len=100,
pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$'),
'password': create_validator('Password', min_len=8, max_len=128)
}
# Проверяем данные формы
form_data = {
'username': 'ab', # Слишком короткий
'email': 'test@example', # Неверный формат
'password': 'secure_pass_123'
}
for field, value in form_data.items():
errors = validators[field](value)
if errors:
for error in errors:
print(error) |
|
Колбэки с захваченным контекстом
В асинхронном коде постоянно нужны колбэки. Проблема: как передать в колбэк дополнительные данные, кроме результата операции? Замыкания решают это элегантно. Реализовывал систему обработки загрузок файлов. Когда файл загружается, нужно обновить запись в БД и отправить уведомление. ID записи и email пользователя известны при создании обработчика:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
| def create_upload_handler(file_id, user_email, db_conn):
"""Создаёт обработчик завершения загрузки"""
start_time = time.time()
def on_complete(file_path):
elapsed = time.time() - start_time
# Обновляем БД - file_id захвачен
db_conn.execute(
"UPDATE files SET status='completed', duration=? WHERE id=?",
(elapsed, file_id)
)
# Отправляем email - user_email захвачен
send_email(
to=user_email,
subject="Файл загружен",
body=f"Файл {file_path} обработан за {elapsed:.2f}с"
)
print(f"Файл {file_id} обработан для {user_email}")
return on_complete
# При запуске загрузки
handler = create_upload_handler(
file_id=12345,
user_email='user@example.com',
db_conn=database
)
# Передаем колбэк в асинхронную систему
async_upload_system.register_callback(handler) |
|
Генераторы с сохранением состояния
Генераторы сами по себе сохраняют состояние через yield. Но иногда нужна дополнительная логика между вызовами - счетчики, аккумуляторы, условия останова. Замыкание оборачивает генератор, добавляя нужную функциональность.
Нужен был генератор, который останавливается после N ошибок или по таймауту:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
| def resilient_stream(source_generator, max_errors=3, timeout=30):
"""Обёртка для генератора с контролем ошибок и таймаута"""
error_count = 0
start_time = time.time()
def generator():
nonlocal error_count
for item in source_generator:
# Проверка таймаута
if time.time() - start_time > timeout:
print(f"Таймаут {timeout}с исчерпан")
break
try:
yield item
error_count = 0 # Сбрасываем счетчик при успехе
except Exception as e:
error_count += 1
print(f"Ошибка {error_count}/{max_errors}: {e}")
if error_count >= max_errors:
print("Превышен лимит ошибок")
break
return generator()
# Использование
def flaky_source():
"""Генератор с периодическими ошибками"""
for i in range(100):
if i % 7 == 0 and i > 0:
raise ValueError(f"Ошибка на элементе {i}")
yield i
stream = resilient_stream(flaky_source(), max_errors=2, timeout=5)
for value in stream:
try:
# Обработка
pass
except ValueError:
continue |
|
error_count и start_time живёт на протяжении всей итерации. Можно создать несколько потоков с разными лимитами - каждый со своим состоянием.
Эти паттерны покрывают процентов восемьдесят случаев, когда замыкания действительно уместны. Остальные двадцать - либо переусложнение, либо случаи где класс подошёл бы лучше. Главное - понимать механику и не применять замыкания везде подряд просто потому что "это функциональное программирование".
Инкапсуляция через замыкания вместо классов
Python не имеет настоящих приватных переменных - всё публично, соглашение с подчеркиванием _private лишь намёк. Но замыкания позволяют создать действительно недоступные извне данные. Захваченные переменные существуют только внутри функции, снаружи к ним не добраться. Делал API-клиент для платежной системы. Токен доступа нужно хранить в памяти, но так чтобы его нельзя было случайно утечь через сериализацию или отладочный вывод. Класс с __dict__ или __slots__ всё равно даёт доступ через интроспекцию. Замыкание - другое дело:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
| def create_api_client(api_key, secret):
"""Создаёт клиент с защищённым токеном"""
# api_key и secret недоступны извне - только через методы
session_token = None
request_count = 0
def authenticate():
nonlocal session_token, request_count
# В реальности - запрос к API
session_token = f"token_{hash((api_key, secret))}"
request_count = 0
return True
def make_request(endpoint, data):
nonlocal request_count
if session_token is None:
authenticate()
request_count += 1
# Используем session_token - он скрыт от внешнего мира
headers = {'Authorization': f'Bearer {session_token}'}
# ... запрос к API ...
return {'status': 'ok', 'endpoint': endpoint}
def get_stats():
# Возвращаем статистику без секретных данных
return {
'requests': request_count,
'authenticated': session_token is not None
}
# Возвращаем интерфейс без доступа к секретам
return {
'request': make_request,
'auth': authenticate,
'stats': get_stats
}
client = create_api_client('key_123', 'secret_xyz')
client['auth']()
result = client['request']('/users', {})
print(client['stats']()) # {'requests': 1, 'authenticated': True}
# Попробуйте найти api_key - не получится
# client.__dict__ - пусто, это словарь с функциями
# dir(client) - только ключи словаря |
|
__closure__. Для класса достаточно vars(obj) или obj.__dict__.
Правда, защита не абсолютная. Если очень постараться, через gc.get_objects() можно найти все ячейки и перебрать. Но от случайной утечки через логирование или сериализацию защищает отлично. Когда писал систему с PCI DSS требованиями, аудиторы оценили - данные карт нигде не торчали в памяти открыто.
Системы подписок и событийные шины
Классическая задача: несколько компонентов должны реагировать на события. Можно сделать паттерн Observer через классы, а можно через замыкания - легче и гибче. Реализовывал систему мониторинга для микросервисов. Каждый сервис отправляет метрики, несколько подписчиков обрабатывают их по-разному: один пишет в базу, другой отправляет алерты, третий обновляет дашборд. Подписчики добавляются и удаляются динамически.
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
| def create_event_bus():
"""Создаёт шину событий с подписками"""
subscribers = {} # topic -> список колбэков
event_history = [] # Последние N событий
def subscribe(topic, handler, filter_fn=None):
"""Подписаться на топик с опциональным фильтром"""
if topic not in subscribers:
subscribers[topic] = []
# Создаём замыкание-обёртку с фильтром
def filtered_handler(event):
if filter_fn is None or filter_fn(event):
handler(event)
subscribers[topic].append(filtered_handler)
# Возвращаем функцию отписки
def unsubscribe():
if topic in subscribers:
subscribers[topic].remove(filtered_handler)
return unsubscribe
def publish(topic, event):
"""Публикует событие всем подписчикам"""
event_history.append((topic, event))
if len(event_history) > 100:
event_history.pop(0)
if topic not in subscribers:
return 0
delivered = 0
for handler in subscribers[topic]:
try:
handler(event)
delivered += 1
except Exception as e:
print(f"Ошибка в обработчике: {e}")
return delivered
def get_history(topic=None, limit=10):
"""Возвращает историю событий"""
if topic:
filtered = [(t, e) for t, e in event_history if t == topic]
return filtered[-limit:]
return event_history[-limit:]
return {
'subscribe': subscribe,
'publish': publish,
'history': get_history
}
# Использование
bus = create_event_bus()
# Подписчик 1: логирование всех метрик
unsubscribe_logger = bus['subscribe'](
'metrics',
lambda event: print(f"LOG: {event}")
)
# Подписчик 2: алерты только для критических метрик
def alert_handler(event):
print(f"ALERT: {event['service']} - {event['value']}")
unsubscribe_alerts = bus['subscribe'](
'metrics',
alert_handler,
filter_fn=lambda e: e.get('level') == 'critical'
)
# Публикация событий
bus['publish']('metrics', {'service': 'api', 'value': 42, 'level': 'info'})
[H2]LOG: {'service': 'api', 'value': 42, 'level': 'info'}[/H2]
bus['publish']('metrics', {'service': 'db', 'value': 95, 'level': 'critical'})
# LOG: {'service': 'db', 'value': 95, 'level': 'critical'}
[H2]ALERT: db - 95[/H2]
# Отписка
unsubscribe_logger()
bus['publish']('metrics', {'service': 'cache', 'value': 10, 'level': 'info'})
# Логгер больше не получает события, только алерты (если критично) |
|
Фильтры тоже реализованы через замыкания - оборачиваем исходный обработчик, добавляем проверку. Можно создавать сложные комбинации:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
| # Обработчик с ограничением частоты
def rate_limited_subscription(topic, handler, max_per_second):
last_calls = []
bus_instance = bus
def rate_limited_handler(event):
import time
now = time.time()
# Чистим старые вызовы
nonlocal last_calls
last_calls = [t for t in last_calls if now - t < 1.0]
if len(last_calls) >= max_per_second:
print(f"Rate limit: пропущено событие")
return
last_calls.append(now)
handler(event)
return bus_instance['subscribe'](topic, rate_limited_handler)
# Подписка с лимитом 2 события в секунду
unsubscribe = rate_limited_subscription(
'metrics',
lambda e: print(f"Processed: {e}"),
max_per_second=2
) |
|
last_calls хранит временные метки. Замыкание инкапсулирует логику ограничения частоты - не нужен отдельный класс или глобальное состояние.
В продакшене такая шина событий обрабатывала тысячи событий в секунду. Единственная проблема - при большом количестве подписчиков (сотни) начинаются тормоза. Решил переписав на слабые ссылки - подписчики удаляются автоматически когда их больше никто не использует:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
| import weakref
def subscribe_weak(topic, handler):
# Создаём слабую ссылку на обработчик
weak_handler = weakref.ref(handler)
def wrapper(event):
h = weak_handler()
if h is not None: # Обработчик еще жив
h(event)
else:
# Автоматически отписываемся
if wrapper in subscribers.get(topic, []):
subscribers[topic].remove(wrapper)
subscribers[topic].append(wrapper) |
|
Распространенные ошибки и их решение
Три года копания в чужом коде научили меня: большинство багов с замыканиями повторяются из проекта в проект. И самое обидное - эти ошибки не ловятся линтерами, тесты проходят, а в продакшене начинается магия. В банковском проекте однажды обнаружил, что все обработчики транзакций использовали один и тот же ID клиента. Оказалось - классический случай позднего связывания в цикле.
Позднее связывание в циклах
Когда создаешь функции в цикле, интуитивно ожидаешь что каждая функция запомнит своё значение счётчика. Python так не работает - все функции захватывают ссылку на ПЕРЕМЕННУЮ, а не значение. Когда цикл завершается, переменная содержит последнее значение:
| Python | 1
2
3
4
5
6
7
8
9
| # Типичная ошибка
handlers = []
for i in range(5):
def handler():
return i # Захватывает переменную i, не значение
handlers.append(handler)
# Все функции возвращают 4!
print([h() for h in handlers]) # [4, 4, 4, 4, 4] |
|
Первое - параметр по умолчанию. Значение параметра по умолчанию вычисляется в момент определения функции:
| Python | 1
2
3
4
5
6
7
| handlers = []
for i in range(5):
def handler(x=i): # i вычисляется СРАЗУ
return x
handlers.append(handler)
print([h() for h in handlers]) # [0, 1, 2, 3, 4] |
|
x никто извне не передает, он существует только для захвата значения. Чтение кода усложняется - не сразу понятно зачем этот параметр.
Второе - создать фабрику функций. Дополнительный уровень вложенности, но семантика чище:
| Python | 1
2
3
4
5
6
7
| def make_handler(value):
def handler():
return value # Захватывает параметр make_handler
return handler
handlers = [make_handler(i) for i in range(5)]
print([h() for h in handlers]) # [0, 1, 2, 3, 4] |
|
make_handler создает новую область видимости с собственным value. Внутренняя функция захватывает этот параметр. Мне такой подход нравится больше - явно видно намерение создать замыкание.
Третье - функциональный стиль с partial:
| Python | 1
2
3
4
5
6
7
| from functools import partial
def handler(value):
return value
handlers = [partial(handler, i) for i in range(5)]
print([h() for h in handlers]) # [0, 1, 2, 3, 4] |
|
partial создает новую функцию с частично примененными аргументами. Но результат тот же, а код компактнее. Правда, если нужна сложная логика внутри обработчика, придется всё равно делать замыкание.
Изменение захваченных переменных без nonlocal
Забыл nonlocal - получил UnboundLocalError или, что хуже, создал локальную переменную вместо изменения внешней. Это ловушка, в которую попадал раз десять за год.
| Python | 1
2
3
4
5
6
7
8
| def counter():
count = 0
def increment():
count = count + 1 # UnboundLocalError!
return count
return increment |
|
count = ... и решает что count локальная. Но в правой части count + 1 пытается прочитать эту локальную переменную, которая еще не инициализирована. Бац - ошибка.
Правильно:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
| def counter():
count = 0
def increment():
nonlocal count # Явно указываем: изменяем внешнюю переменную
count += 1
return count
return increment
cnt = counter()
print(cnt()) # 1
print(cnt()) # 2 |
|
nonlocal:
| Python | 1
2
3
4
5
6
7
8
| def counter():
state = {'count': 0} # Изменяемый объект
def increment():
state['count'] += 1 # Работает! Не переприсваиваем state
return state['count']
return increment |
|
state, а изменяем объект, на который она ссылается. Python различает эти операции. Некоторые используют это как трюк для обхода nonlocal в Python 2, где nonlocal не было. Я считаю это антипаттерном - неявное поведение, которое запутывает.
Однажды рефакторил чужой код, где счетчик хранился в одноэлементном списке: counter = [0]. Автор пытался избежать nonlocal, но получилось хуже. Код работал, но выглядел как шифровка. Переписал с нормальным nonlocal - читабельность выросла в разы.
Мутация разделяемого состояния в многопоточности
Замыкания создают иллюзию потокобезопасности. Кажется: "Каждое замыкание имеет своё состояние, значит конфликтов не будет". Но если несколько потоков вызывают ОДНО замыкание, изменяющее захваченную переменную - привет, гонки данных.
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
| import threading
def create_counter():
count = 0
def increment():
nonlocal count
temp = count # Читаем
# Контекстное переключение может произойти здесь
count = temp + 1 # Записываем
return count
return increment
counter = create_counter()
def worker():
for _ in range(1000):
counter()
threads = [threading.Thread(target=worker) for _ in range(10)]
for t in threads: t.start()
for t in threads: t.join()
# Ожидаем 10000, получаем меньше (например, 8734) |
|
count += 1 не атомарна. Она разбивается на чтение, инкремент и запись. Между этими шагами может произойти переключение контекста. Два потока читают одно значение, инкрементируют, записывают - одно обновление потеряно.
Решение - блокировка:
| Python | 1
2
3
4
5
6
7
8
9
10
11
| def create_thread_safe_counter():
count = 0
lock = threading.Lock()
def increment():
nonlocal count
with lock:
count += 1
return count
return increment |
|
Утечки памяти через циклические ссылки
Замыкание ссылается на объект, объект хранит ссылку на замыкание - цикл. Сборщик мусора Python справляется с циклами, но не мгновенно. В долгоживущих приложениях это накапливается.
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
| class Container:
def __init__(self):
self.items = []
def add_item(item):
self.items.append(item) # Замыкание ссылается на self
self.add = add_item # self ссылается на замыкание
# Цикл: self -> add_item -> self
container = Container()
# Даже после del container объект не освободится сразу |
|
Решил разорвав циклы - использовал слабые ссылки (weakref) там где объект должен был очищаться автоматически:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
| import weakref
class Container:
def __init__(self):
self.items = []
weak_self = weakref.ref(self)
def add_item(item):
strong_self = weak_self()
if strong_self is not None:
strong_self.items.append(item)
self.add = add_item |
|
None - слабая ссылка может протухнуть в любой момент.
Альтернативы замыканиям
Замыкания - не единственный способ упаковать данные и поведение вместе. Иногда они даже не лучший. Год назад рефакторил кодовую базу, где автор использовал замыкания везде - для валидаторов, для конфигурации, для состояния. Код работал, но понимание занимало время. Профилирование показало что в горячих участках классы были быстрее процентов на пятнадцать.
Классы: когда структура важнее краткости
Классический выбор между замыканием и классом сводится к балансу между компактностью и явностью. Замыкание короче, класс - прозрачнее. Но есть конкретные ситуации где класс побеждает безусловно.
Первое - когда нужно несколько методов. Замыкание возвращает одну функцию (или словарь функций). Класс даёт полноценный интерфейс:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
| # Замыкание - громоздко при множестве операций
def create_calculator(initial=0):
value = initial
def add(x):
nonlocal value
value += x
return value
def subtract(x):
nonlocal value
value -= x
return value
def reset():
nonlocal value
value = initial
return {'add': add, 'subtract': subtract, 'reset': reset}
# Класс - естественнее
class Calculator:
def __init__(self, initial=0):
self._initial = initial
self.value = initial
def add(self, x):
self.value += x
return self.value
def subtract(self, x):
self.value -= x
return self.value
def reset(self):
self.value = self._initial
calc = Calculator(10)
calc.add(5)
calc.subtract(3) |
|
dir(), нормальное отображение в отладчике. Замыкание выглядит как словарь - нужно помнить какие ключи доступны.
Второе - наследование и композиция. Попробуй расширить функциональность замыкания. Придется переписывать фабрику или городить wrapper. С классом - просто наследуйся:
| Python | 1
2
3
4
5
6
7
8
| class LoggingCalculator(Calculator):
def add(self, x):
print(f"Добавление {x}")
return super().add(x)
def subtract(self, x):
print(f"Вычитание {x}")
return super().subtract(x) |
|
Третье - сериализация. Замыкания не сериализуются стандартными средствами. Pickle не умеет, JSON тем более. Класс с __dict__ или __slots__ сериализуется тривиально:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| import json
class Config:
def __init__(self, host, port, timeout=30):
self.host = host
self.port = port
self.timeout = timeout
def to_json(self):
return json.dumps(self.__dict__)
@classmethod
def from_json(cls, data):
return cls(**json.loads(data))
cfg = Config('localhost', 8080)
serialized = cfg.to_json() # {"host": "localhost", "port": 8080, "timeout": 30} |
|
dill - работает, но хрупко. Изменишь версию библиотеки или структуру кода - десериализация падает.
Производительность: измеряем на практике
Когда оптимизировал обработку транзакций (обрабатывали миллионы в день), провел бенчмарки. Три варианта хранения состояния: замыкание, класс, dataclass.
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
| import time
# Вариант 1: Замыкание
def create_counter_closure():
count = 0
def increment():
nonlocal count
count += 1
return count
return increment
# Вариант 2: Обычный класс
class CounterClass:
def __init__(self):
self.count = 0
def increment(self):
self.count += 1
return self.count
# Вариант 3: Класс со __slots__
class CounterSlots:
__slots__ = ['count']
def __init__(self):
self.count = 0
def increment(self):
self.count += 1
return self.count
# Вариант 4: Dataclass (Python 3.7+)
from dataclasses import dataclass
@dataclass
class CounterData:
count: int = 0
def increment(self):
self.count += 1
return self.count
# Бенчмарк
iterations = 1_000_000
# Замыкание
closure = create_counter_closure()
start = time.perf_counter()
for _ in range(iterations):
closure()
time_closure = time.perf_counter() - start
# Класс
obj = CounterClass()
start = time.perf_counter()
for _ in range(iterations):
obj.increment()
time_class = time.perf_counter() - start
# Slots
obj_slots = CounterSlots()
start = time.perf_counter()
for _ in range(iterations):
obj_slots.increment()
time_slots = time.perf_counter() - start
# Dataclass
obj_data = CounterData()
start = time.perf_counter()
for _ in range(iterations):
obj_data.increment()
time_data = time.perf_counter() - start
print(f"Замыкание: {time_closure:.3f}s")
print(f"Класс: {time_class:.3f}s")
print(f"__slots__: {time_slots:.3f}s")
print(f"Dataclass: {time_data:.3f}s") |
|
Замыкание: 0.087s
Класс: 0.076s (быстрее на ~13%)
__slots__: 0.068s (быстрее на ~22%)
Dataclass: 0.074s (быстрее на ~15%)
Удивило? Обычный класс оказался быстрее замыкания. Причина - LOAD_DEREF медленнее LOAD_FAST. Доступ к атрибутам через self.count оптимизирован лучше чем разыменование ячейки замыкания. А __slots__ вообще убивает __dict__, снижая overhead.
Но это микробенчмарк. В реальном коде разница нивелируется другими операциями. Когда замерял на обработке транзакций (парсинг JSON, валидация, запись в БД), разница между замыканием и классом составила меньше 1%. Не стоит оптимизировать преждевременно.
Functools.partial: простота без магии
Иногда не нужно полноценное замыкание - достаточно зафиксировать часть аргументов функции. partial делает именно это без дополнительных уровней вложенности:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
| from functools import partial
def send_email(to, subject, body, from_addr='noreply@example.com'):
print(f"От: {from_addr}")
print(f"Кому: {to}")
print(f"Тема: {subject}")
print(f"Текст: {body}")
# Вместо замыкания
def create_notifier(from_addr):
def notify(to, subject, body):
send_email(to, subject, body, from_addr)
return notify
# Используем partial
notify_support = partial(send_email, from_addr='support@example.com')
notify_sales = partial(send_email, from_addr='sales@example.com')
notify_support('user@test.com', 'Вопрос', 'Помогите разобраться')
# От: support@example.com
# Кому: user@test.com
# ... |
|
functools.partial. Но поведение похоже и код проще.
Ограничения есть. Partial не позволяет добавить сложную логику - только фиксирует аргументы. Если нужна обработка, условия, изменение состояния - придётся делать полноценное замыкание или класс:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| # Это partial не умеет
def create_retry_caller(func, max_attempts=3):
attempts = 0
def caller(*args, **kwargs):
nonlocal attempts
for i in range(max_attempts):
try:
result = func(*args, **kwargs)
attempts += 1
return result
except Exception as e:
if i == max_attempts - 1:
raise
print(f"Попытка {i+1} провалилась: {e}")
return caller |
|
Протоколы и структурная типизация
Python 3.8 добавил протоколы из модуля typing. Это способ определить интерфейс без явного наследования - структурная типизация вместо номинальной. Можно использовать вместо замыканий где важна типобезопасность:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
| from typing import Protocol
class Validator(Protocol):
def validate(self, value: str) -> bool:
...
# Любой объект с методом validate(str) -> bool подходит
class EmailValidator:
def validate(self, value: str) -> bool:
return '@' in value and '.' in value
def create_email_validator():
# Замыкание - работает, но без типизации
def validate(value: str) -> bool:
return '@' in value and '.' in value
return validate
# Функция принимает любой Validator
def check_input(validator: Validator, value: str) -> None:
if validator.validate(value):
print("Валидно")
else:
print("Невалидно")
# Работает и с классом, и с замыканием
class_validator = EmailValidator()
closure_validator = create_email_validator()
check_input(class_validator, 'test@example.com') # OK
check_input(closure_validator, 'test@example.com') # Тоже OK |
|
В проекте с жесткими требованиями к типобезопасности (медицинская система) предпочли классы с явными протоколами. Замыкания использовали только для простых callback'ов. Решение спорное, но типизация реально ловила баги на этапе разработки.
Асинхронные замыкания и захват контекста
С async/await замыкания становятся интереснее. Корутина может захватывать переменные из окружающей области, но выполнение откладывается. Пока корутина ждёт, захваченные переменные могут измениться - получается тот же эффект что и в циклах.
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| import asyncio
async def process_items():
tasks = []
for i in range(5):
async def task():
await asyncio.sleep(0.1)
return i # Захватывает i
tasks.append(task())
results = await asyncio.gather(*tasks)
return results
# Все результаты будут 4!
results = asyncio.run(process_items())
print(results) # [4, 4, 4, 4, 4] |
|
i становится 4. Когда корутины выполняются - читают текущее значение. Решение то же - зафиксировать через параметр или создать фабрику:
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
| async def process_items_fixed():
async def make_task(value):
async def task():
await asyncio.sleep(0.1)
return value
return task()
tasks = [make_task(i) for i in range(5)]
return await asyncio.gather(*tasks)
results = asyncio.run(process_items_fixed())
print(results) # [0, 1, 2, 3, 4] |
|
Демонстрационное приложение TaskFlow
Собрал все техники из статьи в одно приложение - систему управления задачами TaskFlow. В реальном проекте такое вряд ли соберешь в одном месте, но для демонстрации возможностей замыканий подходит идеально. Использовал замыкания для кэширования запросов, управления состоянием задач, создания обработчиков событий и декораторов.
| Python | 1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
| import time
from datetime import datetime
from typing import Callable, Dict, List, Any
from functools import wraps
class TaskFlow:
"""Система управления задачами с замыканиями"""
def __init__(self):
self._tasks: Dict[int, Dict] = {}
self._next_id = 1
self._subscribers: Dict[str, List[Callable]] = {}
def create_task_manager(self, user_id: int):
"""Фабрика: создаёт менеджер задач для конкретного пользователя"""
user_cache = {'last_fetch': 0, 'data': None}
stats = {'created': 0, 'completed': 0, 'failed': 0}
def get_tasks(force_refresh=False):
"""Получить задачи с кэшированием"""
nonlocal user_cache
now = time.time()
# Кэш действует 30 секунд
if not force_refresh and user_cache['data'] and now - user_cache['last_fetch'] < 30:
print(f"[Cache] Возврат из кэша для user {user_id}")
return user_cache['data']
print(f"[DB] Загрузка задач для user {user_id}")
# Фильтруем задачи пользователя
user_tasks = {tid: task for tid, task in self._tasks.items()
if task['user_id'] == user_id}
user_cache['data'] = user_tasks
user_cache['last_fetch'] = now
return user_tasks
def create_task(self, title: str, priority: int = 1):
"""Создать новую задачу"""
nonlocal stats
task_id = self._next_id
self._next_id += 1
task = {
'id': task_id,
'user_id': user_id,
'title': title,
'priority': priority,
'status': 'pending',
'created_at': datetime.now()
}
self._tasks[task_id] = task
stats['created'] += 1
# Инвалидируем кэш
user_cache['data'] = None
# Оповещаем подписчиков
self._notify('task_created', task)
return task_id
def complete_task(self, task_id: int):
"""Отметить задачу выполненной"""
nonlocal stats
if task_id in self._tasks and self._tasks[task_id]['user_id'] == user_id:
self._tasks[task_id]['status'] = 'completed'
stats['completed'] += 1
user_cache['data'] = None # Инвалидация
self._notify('task_completed', self._tasks[task_id])
return True
return False
def get_stats(self):
"""Статистика пользователя"""
return stats.copy()
# Возвращаем интерфейс
return {
'get_tasks': get_tasks,
'create': create_task,
'complete': complete_task,
'stats': get_stats
}
def monitored(self, operation: str):
"""Декоратор для мониторинга операций"""
def decorator(func):
call_times = []
@wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
try:
result = func(*args, **kwargs)
elapsed = time.perf_counter() - start
call_times.append(elapsed)
# Храним только последние 100 вызовов
if len(call_times) > 100:
call_times.pop(0)
avg_time = sum(call_times) / len(call_times)
print(f"[Monitor] {operation}: {elapsed:.4f}s (avg: {avg_time:.4f}s)")
return result
except Exception as e:
print(f"[Monitor] {operation} failed: {e}")
raise
wrapper.get_metrics = lambda: {
'calls': len(call_times),
'avg_time': sum(call_times) / len(call_times) if call_times else 0,
'last_times': call_times[-10:]
}
return wrapper
return decorator
def subscribe(self, event: str, handler: Callable):
"""Подписка на события с замыканием-wrapper'ом"""
if event not in self._subscribers:
self._subscribers[event] = []
# Создаем замыкание с контекстом подписки
subscription_time = datetime.now()
call_count = {'value': 0}
def wrapped_handler(data):
call_count['value'] += 1
try:
handler(data)
except Exception as e:
print(f"Ошибка в обработчике {event}: {e}")
# Добавляем метаданные
wrapped_handler.event = event
wrapped_handler.subscribed_at = subscription_time
wrapped_handler.call_count = lambda: call_count['value']
self._subscribers[event].append(wrapped_handler)
# Возвращаем функцию отписки
def unsubscribe():
if wrapped_handler in self._subscribers.get(event, []):
self._subscribers[event].remove(wrapped_handler)
return unsubscribe
def _notify(self, event: str, data: Any):
"""Оповестить подписчиков о событии"""
for handler in self._subscribers.get(event, []):
handler(data)
# Демонстрация работы
system = TaskFlow()
# Создаём менеджеров для двух пользователей
alice_mgr = system.create_task_manager(user_id=1)
bob_mgr = system.create_task_manager(user_id=2)
# Подписываемся на события
unsubscribe_logger = system.subscribe(
'task_created',
lambda task: print(f"✓ Создана задача: {task['title']}")
)
unsubscribe_stats = system.subscribe(
'task_completed',
lambda task: print(f"✓ Выполнена задача #{task['id']}")
)
# Работа с задачами Alice
print("=== Alice создаёт задачи ===")
alice_mgr['create']("Написать отчет", priority=2)
alice_mgr['create']("Код-ревью PR#123", priority=3)
print("\n=== Alice получает задачи (первый раз) ===")
tasks = alice_mgr['get_tasks']()
print(f"Найдено задач: {len(tasks)}")
print("\n=== Alice получает задачи (из кэша) ===")
tasks = alice_mgr['get_tasks']()
print("\n=== Bob создаёт свою задачу ===")
bob_mgr['create']("Обновить документацию")
print("\n=== Alice завершает задачу ===")
task_id = list(alice_mgr['get_tasks']().keys())[0]
alice_mgr['complete'](task_id)
print("\n=== Статистика ===")
print(f"Alice: {alice_mgr['stats']()}")
print(f"Bob: {bob_mgr['stats']()}")
# Отписываемся
unsubscribe_logger()
print("\n=== После отписки логгера ===")
alice_mgr['create']("Ещё одна задача") |
|
monitored (не использовал в примере, но он готов) замеряет время выполнения и хранит метрики в захваченном списке. Система подписок создаёт wrapper'ы-замыкания которые добавляют метаданные и обработку ошибок.
Запустил этот код - работает как часы. Кэш инвалидируется при создании и завершении задач. События доставляются всем подписчикам. Каждый пользователь получает собственный менеджер с изолированным состоянием. Всё это на замыканиях, без единого класса для бизнес-логики (TaskFlow - просто контейнер).
Используя замыкания функций, объявите внутреннюю функцию, которая принимает в качестве параметров фамилию и имя
1) Используя замыкания функций, объявите внутреннюю функцию, которая принимает в качестве... Вызовите внутреннюю функцию замыкания и отобразите на экране результат ее работы
Используя замыкания функций, объявите внутреннюю функцию, которая принимает два
параметра a , b ,... Вызовите внутреннюю функцию замыкания и отобразите на экране результат ее работы
Используя замыкания функций, объявите внутреннюю функцию, которая принимает два
параметра a , b ,... Использование замыкания
Используя замыкания функций, объявите внутреннюю функцию, которая принимает в
качестве аргумента... Используя замыкания функций, объявите внутреннюю функцию, которая преобразует строку из списка целых чисел
Используя замыкания функций, объявите внутреннюю функцию, которая преобразует строку
из списка... Вызовите внутреннюю функцию замыкания и отобразите на экране результат ее работы
Используя замыкания функций, объявите внутреннюю функцию, которая принимает два
параметра a , b ,... Зачем изучать Python как веб-разработчик?
Это может быть глупый вопрос, но я задаю его прямо:
Сегодня, как веб-разработчик с Node на... Зачем биты нужны это меньше байтов но int 32 бита но я не допер зачем это нужно это 4 байта то есть int не может больше 4 байт весить?
Вот еще один вопрос зачем биты нужны это меньше байтов но int 32 бита но я не допер зачем это... Замыкания, и замыкания в объекте. Где данные?
Помогите пожалуйста разобраться где хранятся данные. Ниже 2 примера. В первом все вроде как понятно... Какие потери холостого хода; потери короткого замыкания; напряжение короткого замыкания;ток холостого хода у трансформатора ТСЗИ-5,0 на напряжение 380
Можете пожалуйста подсказать какие потери холостого хода; потери короткого замыкания; напряжение... Захват элемента, узнать в текущий момент имеется ли захват
let r = document.querySelector('div');
document.onpointerdown = e => {
... Что дает сравнение указателей, зачем это нужно и как это можно использовать
Подскажите пожалуйста. Что дает сравнение указателей, зачем это нужно и как это можно использовать...
|
