Замыкания в Python

def приветствие(имя):
    return f"Привет, {имя}!"
 
# Функцию можно присвоить переменной
поздороваться = приветствие
 
# Теперь можно вызвать функцию через новое имя
print(поздороваться("Андрей"))  # Выведет: Привет, Андрей!
 
# Функцию можно передать как аргумент
def выполнить_функцию(функция, аргумент):
    return функция(аргумент)
 
print(выполнить_функцию(приветствие, "Елена"))  # Привет, Елена!
 
# Функцию можно вернуть из другой функции
def получить_функцию_приветствия():
    return приветствие
 
новое_приветствие = получить_функцию_приветствия()
print(новое_приветствие("Сергей"))  # Привет, Сергей!

def внешняя_функция(x):
    # Переменная внешней функции
    сообщение = f"Значение x: {x}"
    
    def внутренняя_функция():
        # Внутренняя функция имеет доступ к переменным внешней
        print(сообщение)
    
    # Вызываем внутреннюю функцию
    внутренняя_функция()
 
внешняя_функция(10)  # Выведет: Значение x: 10

x = "глобальная"
 
def внешняя():
    x = "охватывающая"
    
    def внутренняя():
        x = "локальная"
        print(f"Внутри внутренней: {x}")
    
    внутренняя()
    print(f"Внутри внешней: {x}")
 
внешняя()
print(f"В глобальной области: {x}")

Внутри внутренней: локальная
Внутри внешней: охватывающая
В глобальной области: глобальная

def создать_умножитель(множитель):
    def умножитель(число):
        return число * множитель  # множитель из внешней функции
    return умножитель
 
удвоить = создать_умножитель(2)
утроить = создать_умножитель(3)
 
print(удвоить(5))  # 10
print(утроить(5))  # 15

def создать_счётчик():
    счётчик = 0
    
    def увеличить():
        nonlocal счётчик  # Объявляем, что используем переменную из внешней области
        счётчик += 1
        return счётчик
    
    return увеличить
 
мой_счётчик = создать_счётчик()
print(мой_счётчик())  # 1
print(мой_счётчик())  # 2
print(мой_счётчик())  # 3

def проблемный_счётчик():
    счётчик = 0
    
    def увеличить():
        # Без nonlocal будет ошибка: UnboundLocalError
        счётчик += 1  # Python считает, что счётчик - локальная переменная
        return счётчик
    
    return увеличить
 
# Вызов этой функции приведёт к ошибке

# Обычная функция
def simple_function(x):
    return x * 2
 
# Функция, возвращающая замыкание
def make_multiplier(factor):
    def multiplier(x):
        return x * factor  # использует переменную из внешней области
    return multiplier

def outer_function():
    x = 10
    def inner_function():
        print(x)  # x из окружения outer_function
    return inner_function
 
closure = outer_function()
closure()  # Выведет: 10

closure.__closure__[0].cell_contents  # Получим 10

def counter_factory():
    count = 0
    def increment():
        nonlocal count  # указываем, что хотим изменить переменную из внешней области
        count += 1
        return count
    return increment
 
counter = counter_factory()
print(counter())  # 1
print(counter())  # 2

def create_functions():
    functions = []
    for i in range(3):
        def func():
            return i
        functions.append(func)
    return functions
 
funcs = create_functions()
[f() for f in funcs]  # [2, 2, 2], а не [0, 1, 2], как можно ожидать

def create_functions_fixed():
    functions = []
    for i in range(3):
        def func(i=i):  # i=i создаёт локальную копию i
            return i
        functions.append(func)
    return functions
 
funcs = create_functions_fixed()
[f() for f in funcs]  # [0, 1, 2] — теперь всё работает как ожидалось

def not_a_closure():
    x = 10
    def inner():
        print("Это просто вложенная функция")
    inner()  # вызываем функцию, а не возвращаем её
    
def also_not_a_closure():
    def inner():
        print("Здесь нет захвата внешних переменных")
    return inner  # возвращаем функцию, но без захвата внешних переменных
    
def this_is_a_closure():
    x = 10
    def inner():
        print(f"Замыкание захватило x = {x}")
    return inner  # возвращаем функцию, которая захватывает x

def power_factory(exponent):
    def power_function(base):
        return base ** exponent
    return power_function
 
square = power_factory(2)
cube = power_factory(3)
 
print(square(4))  # 16
print(cube(4))    # 64

def make_averager():
    series = []
    
    def averager(new_value):
        series.append(new_value)
        return sum(series) / len(series)
    
    return averager
 
avg = make_averager()
print(avg(10))  # 10.0
print(avg(15))  # 12.5
print(avg(20))  # 15.0

def outer(x):
    y = 10
    def inner():
        return x + y
    return inner
 
closure1 = outer(5)
closure2 = outer(10)

# Проверка содержимого замыкания
print(closure1.__closure__[0].cell_contents)  # 5
print(closure1.__closure__[1].cell_contents)  # 10
print(closure2.__closure__[0].cell_contents)  # 10
print(closure2.__closure__[1].cell_contents)  # 10

def create_multipliers():
    return [lambda x: i * x for i in range(5)]

def counter():
    count = 0
    def increment():
        count += 1  # Ошибка! Python считает count локальной переменной
        return count
    return increment

def append_factory():
    data = []
    def appender(item):
        data.append(item)
        return data
    return appender

def outer(x):
    def inner():
        return x
    return inner
 
closure = outer(42)
# Доступ к захваченным переменным
print(closure.__closure__[0].cell_contents)  # 42

import inspect
 
def create_closure(x, y):
    def inner(z):
        return x + y + z
    return inner
 
my_closure = create_closure(10, 20)
 
# Получение исходного кода функции
print(inspect.getsource(my_closure))
[H2]Результат: не исходный код create_closure, а только inner[/H2]
 
# Получение имён свободных переменных в функции
print(my_closure.__code__.co_freevars)  # ('x', 'y')

def debuggable_counter():
    count = 0
    
    def increment():
        nonlocal count
        count += 1
        return count
    
    # Добавляем функцию для отладки
    def debug_get_count():
        return count
    
    increment.debug_get_count = debug_get_count
    return increment
 
counter = debuggable_counter()
print(counter())  # 1
print(counter.debug_get_count())  # 1

def factory(x):
    def closure(y):
        import pdb; pdb.set_trace()  # Точка останова внутри замыкания
        return x + y
    return closure
 
func = factory(10)
result = func(5)  # Здесь активируется отладчик

import weakref
import gc
 
def outer():
    # Создаём большой объект
    big_list = [0] * 1000000
    
    # Создаём слабую ссылку для отслеживания
    weak_ref = weakref.ref(big_list)
    
    def inner():
        # Не используем big_list напрямую
        return "Closure exists"
    
    # Возвращаем замыкание и слабую ссылку
    return inner, weak_ref
 
closure, weak_ref = outer()
 
# Проверяем, существует ли еще большой список
print(weak_ref() is not None)  # True - список еще не собран мусорщиком
 
# Удаляем замыкание
del closure
gc.collect()  # Принудительно запускаем сборку мусора
 
# Теперь список должен быть удален
print(weak_ref() is not None)  # False - список собран мусорщиком

def create_processor(data):
    # Большой набор данных
    processed = [item * 2 for item in data]
    
    def get_item(index):
        return processed[index]
    
    return get_item
 
# Создаём процессор для большого списка
big_data = list(range(1000000))
processor = create_processor(big_data)
 
# big_data можно удалить, так как он больше не нужен
del big_data
 
# Но processed остаётся в памяти из-за замыкания!

def create_efficient_processor(data):
    # Сохраняем только необходимые вычисленные данные
    processed = [item * 2 for item in data]
    
    # Освобождаем оригинальные данные от внешних ссылок
    data = None
    
    def get_item(index):
        return processed[index]
    
    return get_item

def create_circular_reference():
    container = {}
    
    def closure():
        return container
    
    # Создаём циклическую ссылку
    container['self_reference'] = closure
    
    return closure

def cumulative_average():
    values = []
    
    def average(value):
        values.append(value)
        return sum(values) / len(values)
    
    return average
 
avg = cumulative_average()
print(avg(10))  # 10.0
print(avg(20))  # 15.0
print(avg(30))  # 20.0

def make_account(initial_balance=0):
    balance = initial_balance
    
    def deposit(amount):
        nonlocal balance
        balance += amount
        return balance
    
    def withdraw(amount):
        nonlocal balance
        if amount > balance:
            raise ValueError("Недостаточно средств")
        balance -= amount
        return balance
    
    def get_balance():
        return balance
    
    # Возвращаем словарь с доступными операциями
    return {
        'deposit': deposit,
        'withdraw': withdraw,
        'balance': get_balance
    }
 
# Использование
account = make_account(1000)
print(account['balance']())  # 1000
account['deposit'](500)
print(account['balance']())  # 1500
account['withdraw'](200)
print(account['balance']())  # 1300

def make_formatter(prefix, suffix):
    def format_string(string):
        return f"{prefix}{string}{suffix}"
    return format_string
 
html_paragraph = make_formatter("<p>", "</p>")
bold_text = make_formatter("<b>", "</b>")
 
print(html_paragraph("Это параграф"))  # <p>Это параграф</p>
print(bold_text("Жирный текст"))      # <b>Жирный текст</b>

def make_counter(start=0, step=1):
    count = start - step  # Компенсируем первое увеличение
    
    def counter():
        nonlocal count
        count += step
        return count
    
    return counter
 
count_from_ten = make_counter(10)
count_by_twos = make_counter(0, 2)
 
print(count_from_ten())  # 10
print(count_from_ten())  # 11
print(count_from_ten())  # 12
 
print(count_by_twos())   # 0
print(count_by_twos())   # 2
print(count_by_twos())   # 4

def make_accumulator(func=sum):
    items = []
    
    def accumulate(item=None):
        if item is not None:
            items.append(item)
        return func(items)
    
    return accumulate
 
sum_acc = make_accumulator()
max_acc = make_accumulator(max)
 
sum_acc(5)
sum_acc(10)
sum_acc(15)
print(sum_acc())  # 30
 
max_acc(3)
max_acc(7)
max_acc(2)
print(max_acc())  # 7

import tkinter as tk
 
def make_button_handler(message):
    def handler():
        print(f"Нажата кнопка: {message}")
    return handler
 
root = tk.Tk()
root.title("Пример замыканий с GUI")
 
button1 = tk.Button(root, text="Кнопка 1", command=make_button_handler("Первая"))
button2 = tk.Button(root, text="Кнопка 2", command=make_button_handler("Вторая"))
 
button1.pack()
button2.pack()
 
# root.mainloop()  # Раскомментируйте для запуска GUI

import asyncio
 
def create_async_task(task_id):
    async def task():
        print(f"Задача {task_id} начата")
        await asyncio.sleep(1)  # Имитация асинхронной операции
        print(f"Задача {task_id} завершена")
        return f"Результат задачи {task_id}"
    return task
 
# Пример использования
async def main():
    tasks = [create_async_task(i) for i in range(3)]
    results = await asyncio.gather(*(task() for task in tasks))
    print(results)
 
# asyncio.run(main())  # Раскомментируйте для запуска

import asyncio
import random
 
def create_retry_function(max_retries=3, delay=1):
    async def retry_operation(operation, *args, **kwargs):
        retries = 0
        while True:
            try:
                return await operation(*args, **kwargs)
            except Exception as e:
                retries += 1
                if retries > max_retries:
                    raise
                wait_time = delay * (2 ** (retries - 1)) * (0.5 + random.random())
                print(f"Попытка {retries} не удалась. Ожидание {wait_time:.2f}с")
                await asyncio.sleep(wait_time)
    
    return retry_operation
 
# Пример использования
async def fetch_data(url):
    # Имитация запроса, который иногда завершается ошибкой
    if random.random() < 0.7:  # 70% вероятность ошибки
        raise ConnectionError("Ошибка соединения")
    return f"Данные с {url}"
 
async def main():
    retry_fetch = create_retry_function(max_retries=4, delay=0.5)
    try:
        result = await retry_fetch(fetch_data, "https://example.com")
        print(f"Успех: {result}")
    except Exception as e:
        print(f"Окончательная ошибка: {e}")

def create_query_builder(connection_string):
    import sqlite3
    conn = sqlite3.connect(connection_string)
    
    # Кэш для результатов запросов
    query_cache = {}
    
    def execute_query(query, params=None, use_cache=True):
        # Формируем ключ для кэша
        cache_key = (query, str(params)) if params else query
        
        # Проверяем кэш, если разрешено
        if use_cache and cache_key in query_cache:
            print("Используем кэшированный результат")
            return query_cache[cache_key]
        
        # Выполняем запрос
        cursor = conn.cursor()
        if params:
            cursor.execute(query, params)
        else:
            cursor.execute(query)
        
        # Получаем результаты
        results = cursor.fetchall()
        
        # Сохраняем в кэш, если разрешено
        if use_cache:
            query_cache[cache_key] = results
            
        return results
    
    # Добавляем функцию очистки кэша
    def clear_cache():
        nonlocal query_cache
        query_cache = {}
        
    # Привязываем дополнительную функцию к основной
    execute_query.clear_cache = clear_cache
    
    return execute_query
 
# Пример использования
db_query = create_query_builder(":memory:")
 
# Создаем таблицу
db_query("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
 
# Добавляем данные
db_query("INSERT INTO users VALUES (?, ?)", (1, "Иван"))
db_query("INSERT INTO users VALUES (?, ?)", (2, "Мария"))
 
# Первый запрос выполняется и кэшируется
print(db_query("SELECT * FROM users WHERE id=?", (1,)))
 
# Второй идентичный запрос берется из кэша
print(db_query("SELECT * FROM users WHERE id=?", (1,)))
 
# Очищаем кэш
db_query.clear_cache()

def create_counter_app():
    import tkinter as tk
    
    count = 0
    
    # Создаем главное окно
    root = tk.Tk()
    root.title("Счетчик с замыканиями")
    root.geometry("300x150")
    
    # Создаем виджет для отображения счетчика
    label = tk.Label(root, text=str(count), font=("Arial", 24))
    label.pack(pady=20)
    
    # Создаем обработчики для кнопок
    def increment():
        nonlocal count
        count += 1
        label.config(text=str(count))
    
    def decrement():
        nonlocal count
        count -= 1
        label.config(text=str(count))
    
    # Создаем кнопки
    btn_frame = tk.Frame(root)
    btn_frame.pack()
    
    minus_btn = tk.Button(btn_frame, text="-", command=decrement, font=("Arial", 16))
    minus_btn.pack(side=tk.LEFT, padx=10)
    
    plus_btn = tk.Button(btn_frame, text="+", command=increment, font=("Arial", 16))
    plus_btn.pack(side=tk.LEFT, padx=10)
    
    # Возвращаем функцию запуска приложения
    def run():
        root.mainloop()
    
    return run
 
# Создаем и запускаем приложение
# counter_app = create_counter_app()
# counter_app()

def make_multiplier_counter():
    multiplier = 2
    count = 0
    
    def multiply_and_count(n):
        nonlocal count, multiplier
        count += 1
        
        # Периодически увеличиваем множитель
        if count % 10 == 0:
            multiplier += 1
            
        return n * multiplier, count, multiplier
        
    return multiply_and_count
 
counter = make_multiplier_counter()
print(counter(5))   # (10, 1, 2) - число * множитель, счетчик, текущий множитель
print(counter(5))   # (10, 2, 2)

def collection_processor():
    data = []
    
    def process(item):
        # Модификация объекта не требует nonlocal
        data.append(item)
        return data
        
    def clear():
        # Переприсваивание требует nonlocal
        nonlocal data
        data = []
        
    process.clear = clear
    return process
 
processor = collection_processor()
print(processor(1))  # [1]
print(processor(2))  # [1, 2]
processor.clear()
print(processor(3))  # [3]

import time
import functools
 
def timing_decorator(func):
    @functools.wraps(func)  # Сохраняем метаданные оригинальной функции
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        print(f"Функция {func.__name__} выполнилась за {end_time - start_time:.6f} секунд")
        return result
    return wrapper
 
@timing_decorator
def slow_function(delay):
    time.sleep(delay)
    return "Операция завершена"
 
print(slow_function(0.5))

def repeat(n=3):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            results = []
            for _ in range(n):
                results.append(func(*args, **kwargs))
            return results
        return wrapper
    return decorator
 
@repeat(5)
def generate_random():
    import random
    return random.random()
 
print(generate_random())  # Список из 5 случайных чисел

def memoize(func):
    cache = {}
    
    def wrapper(*args):
        if args not in cache:
            cache[args] = func(*args)
        return cache[args]
    
    # Добавим метод для очистки кэша
    def clear_cache():
        nonlocal cache
        cache = {}
        
    wrapper.clear_cache = clear_cache
    return wrapper
 
@memoize
def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)
 
# Без мемоизации функция была бы крайне неэффективной
print(fibonacci(35))  # Почти мгновенный результат благодаря кэшированию

def lazy_loader(expensive_function):
    result = None
    initialized = False
    
    def wrapper(*args, **kwargs):
        nonlocal result, initialized
        if not initialized:
            result = expensive_function(*args, **kwargs)
            initialized = True
        return result
    
    return wrapper
 
@lazy_loader
def load_large_dataset(path="data.csv"):
    print("Загрузка большого набора данных...")
    # В реальности здесь был бы код загрузки данных
    return [i for i in range(1000)]
 
# Данные загружаются только при первом вызове
data = load_large_dataset()
# Последующие вызовы используют кэшированный результат
data_again = load_large_dataset()

def compose(*functions):
    def inner(x):
        result = x
        for func in reversed(functions):
            result = func(result)
        return result
    return inner
 
# Примеры простых функций
def double(x): return x * 2
def square(x): return x ** 2
def increment(x): return x + 1
 
# Композиция функций
pipeline = compose(increment, square, double)
print(pipeline(3))  # (3 * 2)^2 + 1 = 37

def map_reduce(map_func, reduce_func, initial=None):
    def processor(iterable):
        # Применяем map_func к каждому элементу
        mapped = [map_func(item) for item in iterable]
        
        # Применяем reduce_func для свёртки результатов
        if initial is not None:
            result = initial
            for item in mapped:
                result = reduce_func(result, item)
        else:
            result = mapped[0] if mapped else None
            for item in mapped[1:]:
                result = reduce_func(result, item)
                
        return result
    return processor
 
# Вычисление суммы квадратов
sum_of_squares = map_reduce(
    map_func=lambda x: x**2,
    reduce_func=lambda x, y: x + y,
    initial=0
)
 
print(sum_of_squares([1, 2, 3, 4]))  # 30

import time
 
# Реализация с использованием замыкания
def counter_closure():
    count = 0
    def increment():
        nonlocal count
        count += 1
        return count
    return increment
 
# Реализация с использованием класса
class CounterClass:
    def __init__(self):
        self.count = 0
    
    def increment(self):
        self.count += 1
        return self.count
 
# Тестирование производительности
def test_performance(iterations):
    # Тестируем замыкание
    closure_counter = counter_closure()
    start = time.time()
    for _ in range(iterations):
        closure_counter()
    closure_time = time.time() - start
    
    # Тестируем класс
    class_counter = CounterClass()
    start = time.time()
    for _ in range(iterations):
        class_counter.increment()
    class_time = time.time() - start
    
    return closure_time, class_time
 
iterations = 10_000_000
closure_time, class_time = test_performance(iterations)
print(f"Время замыкания: {closure_time:.5f} сек")
print(f"Время класса: {class_time:.5f} сек")
print(f"Соотношение: {closure_time / class_time:.2f}x")

def curry(func):
    def curried(*args):
        if len(args) >= func.__code__.co_argcount:
            return func(*args)
        return lambda *more_args: curried(*(args + more_args))
    return curried
 
@curry
def add3(a, b, c):
    return a + b + c
 
# Можно вызывать по-разному
print(add3(1)(2)(3))     # 6
print(add3(1, 2)(3))     # 6
print(add3(1)(2, 3))     # 6

def partial(func, *fixed_args, **fixed_kwargs):
    def wrapper(*args, [B]kwargs):
        all_kwargs = {[/B]fixed_kwargs, **kwargs}
        return func(*fixed_args, *args, **all_kwargs)
    return wrapper
 
def greet(greeting, name, punctuation='!'):
    return f"{greeting}, {name}{punctuation}"
 
hello = partial(greet, "Hello")
print(hello("World"))  # Hello, World!
 
formal_greeting = partial(greet, punctuation='.')
print(formal_greeting("Good day", "sir"))  # Good day, sir.

def pipeline_builder():
    transformations = []
    
    def add_step(func):
        transformations.append(func)
        return add_step
        
    def build():
        def pipeline(data):
            result = data
            for transform in transformations:
                result = transform(result)
            return result
        return pipeline
        
    add_step.build = build
    return add_step
 
# Создаём конвейер обработки данных
process = (pipeline_builder()
           (lambda x: [i * 2 for i in x])      # Удваиваем каждый элемент
           (lambda x: [i for i in x if i > 5])  # Фильтруем элементы > 5
           (lambda x: sum(x))                  # Суммируем результаты
           .build())
 
data = [1, 2, 3, 4, 5]
print(process(data))  # 28 (2*4 + 2*5 + 2*3 + 2*5 = 8 + 10 + 6 + 4 = 28)

# Альтернатива замыканию с использованием callable класса
class Counter:
    def __init__(self, start=0, step=1):
        self.count = start - step
        self.step = step
        
    def __call__(self):
        self.count += self.step
        return self.count
 
counter = Counter(10)
print(counter())  # 10
print(counter())  # 11

# Python 2
def counter():
    count = [0]  # Изменяемый объект вместо простой переменной
    def increment():
        count[0] += 1
        return count[0]
    return increment