Продвинутые темы

# Функции как объекты первого класса
def greet(name):
    return f"Привет, {name}!"

# Присваивание функции переменной
say_hello = greet
print(say_hello("Иван"))  # Вывод: Привет, Иван!

# Передача функции в качестве аргумента
def execute_function(func, arg):
    return func(arg)

result = execute_function(greet, "Мария")
print(result)  # Вывод: Привет, Мария!

# Возврат функции из другой функции
def get_greeting_function():
    def hello(name):
        return f"Добрый день, {name}!"
    return hello

new_greeting = get_greeting_function()
print(new_greeting("Алексей"))  # Вывод: Добрый день, Алексей!

# Определяем декоратор
def log_function_call(func):
    def wrapper(*args, **kwargs):
        print(f"Вызов функции {func.__name__} с аргументами {args} и {kwargs}")
        result = func(*args, **kwargs)
        print(f"Функция {func.__name__} вернула {result}")
        return result
    return wrapper

# Применяем декоратор
def add_numbers(a, b):
    return a + b

decorated_add = log_function_call(add_numbers)
result = decorated_add(3, 5)
# Вывод:
# Вызов функции add_numbers с аргументами (3, 5) и {}
# Функция add_numbers вернула 8

# Декоратор
def log_function_call(func):
    def wrapper(*args, **kwargs):
        print(f"Вызов функции {func.__name__} с аргументами {args} и {kwargs}")
        result = func(*args, **kwargs)
        print(f"Функция {func.__name__} вернула {result}")
        return result
    return wrapper

# Применение декоратора с синтаксисом @
@log_function_call
def add_numbers(a, b):
    return a + b

# Теперь add_numbers автоматически декорирована
result = add_numbers(3, 5)
# Вывод тот же:
# Вызов функции add_numbers с аргументами (3, 5) и {}
# Функция add_numbers вернула 8

import functools

def log_function_call(func):
    @functools.wraps(func)  # Сохраняет метаданные оригинальной функции
    def wrapper(*args, **kwargs):
        print(f"Вызов функции {func.__name__} с аргументами {args} и {kwargs}")
        result = func(*args, **kwargs)
        print(f"Функция {func.__name__} вернула {result}")
        return result
    return wrapper

@log_function_call
def add_numbers(a, b):
    """Функция складывает два числа и возвращает результат."""
    return a + b

# Теперь метаданные сохранены
print(add_numbers.__name__)  # Вывод: add_numbers
print(add_numbers.__doc__)   # Вывод: Функция складывает два числа и возвращает результат.

import functools

def repeat(times):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            result = None
            for _ in range(times):
                result = func(*args, **kwargs)
            return result
        return wrapper
    return decorator

@repeat(times=3)
def say_hello(name):
    print(f"Привет, {name}!")
    return name

# Эта функция выполнится 3 раза
say_hello("Иван")
# Вывод:
# Привет, Иван!
# Привет, Иван!
# Привет, Иван!

import functools

class CountCalls:
    def __init__(self, func):
        functools.update_wrapper(self, func)
        self.func = func
        self.count = 0
    
    def __call__(self, *args, **kwargs):
        self.count += 1
        print(f"Функция {self.func.__name__} была вызвана {self.count} раз")
        return self.func(*args, **kwargs)

@CountCalls
def say_hello(name):
    return f"Привет, {name}!"

print(say_hello("Иван"))   # Вызов 1
print(say_hello("Мария"))  # Вызов 2
print(say_hello("Петр"))   # Вызов 3

# Вывод:
# Функция say_hello была вызвана 1 раз
# Привет, Иван!
# Функция say_hello была вызвана 2 раз
# Привет, Мария!
# Функция say_hello была вызвана 3 раз
# Привет, Петр!

import functools

def bold(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return f"{func(*args, **kwargs)}"
    return wrapper

def italic(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return f"{func(*args, **kwargs)}"
    return wrapper

@bold
@italic
def greet(name):
    return f"Привет, {name}!"

# Сначала применяется italic, затем bold
print(greet("Иван"))  # Вывод: Привет, Иван!

import time
import functools

def timer(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

@timer
def slow_function():
    time.sleep(1)
    return "Функция выполнена"

slow_function()
# Вывод: Функция slow_function выполнялась 1.001234 секунд

import functools

def enforce_types(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        # Получаем аннотации типов из функции
        annotations = func.__annotations__
        
        # Проверяем типы аргументов
        for arg, arg_name in zip(args, func.__code__.co_varnames):
            if arg_name in annotations and not isinstance(arg, annotations[arg_name]):
                raise TypeError(f"Аргумент {arg_name} должен быть типа {annotations[arg_name].__name__}")
        
        # Проверяем типы именованных аргументов
        for arg_name, arg in kwargs.items():
            if arg_name in annotations and not isinstance(arg, annotations[arg_name]):
                raise TypeError(f"Аргумент {arg_name} должен быть типа {annotations[arg_name].__name__}")
        
        return func(*args, **kwargs)
    return wrapper

@enforce_types
def add_numbers(a: int, b: int) -> int:
    return a + b

print(add_numbers(1, 2))  # OK: 3
try:
    print(add_numbers("1", 2))  # Ошибка: "1" не int
except TypeError as e:
    print(e)  # Вывод: Аргумент a должен быть типа int

import functools

# Для кэширования можно использовать встроенный декоратор lru_cache
@functools.lru_cache(maxsize=128)
def fibonacci(n):
    if n <= 1:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

# Без кэширования эта функция работала бы очень медленно
print(fibonacci(30))  # Быстрый результат благодаря кэшированию

# Создание собственного декоратора кэширования
def cache(func):
    cache_data = {}
    @functools.wraps(func)
    def wrapper(*args):
        if args not in cache_data:
            cache_data[args] = func(*args)
        return cache_data[args]
    return wrapper

@cache
def factorial(n):
    if n <= 1:
        return 1
    return n * factorial(n-1)

print(factorial(10))  # Результат будет кэширован

class Temperature:
    def __init__(self, celsius=0):
        self._celsius = celsius

    @property
    def celsius(self):
        return self._celsius
    
    @celsius.setter
    def celsius(self, value):
        if value < -273.15:
            raise ValueError("Температура не может быть ниже абсолютного нуля")
        self._celsius = value
    
    @property
    def fahrenheit(self):
        return self._celsius * 9/5 + 32
    
    @staticmethod
    def is_valid_temperature(temperature):
        return temperature >= -273.15
    
    @classmethod
    def from_fahrenheit(cls, fahrenheit):
        celsius = (fahrenheit - 32) * 5/9
        return cls(celsius)

# Использование
temp = Temperature(25)
print(temp.celsius)      # Использование геттера: 25
temp.celsius = 30        # Использование сеттера
print(temp.fahrenheit)   # Использование свойства: 86.0

print(Temperature.is_valid_temperature(-300))   # Статический метод: False

# Создание экземпляра через метод класса
temp2 = Temperature.from_fahrenheit(77)
print(temp2.celsius)     # Примерно 25.0

# Пример простого итератора, который возвращает числа от 0 до n-1
class Counter:
    def __init__(self, limit):
        self.limit = limit
        self.counter = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.counter < self.limit:
            value = self.counter
            self.counter += 1
            return value
        else:
            raise StopIteration

# Использование итератора в цикле for
for i in Counter(5):
    print(i)  # Выведет числа от 0 до 4

# Ручное использование итератора
counter = Counter(3)
print(next(counter))  # 0
print(next(counter))  # 1
print(next(counter))  # 2
try:
    print(next(counter))  # Вызовет StopIteration
except StopIteration:
    print("Итератор исчерпан")

# Пример итерируемого объекта (но не итератора)
class NumberSequence:
    def __init__(self, numbers):
        self.numbers = numbers
    
    def __iter__(self):
        # Возвращаем новый итератор для последовательности
        return NumberIterator(self.numbers)

# Итератор для NumberSequence
class NumberIterator:
    def __init__(self, numbers):
        self.numbers = numbers
        self.index = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.index < len(self.numbers):
            value = self.numbers[self.index]
            self.index += 1
            return value
        else:
            raise StopIteration

# Использование
sequence = NumberSequence([1, 2, 3, 4, 5])
for number in sequence:
    print(number)  # Выведет числа от 1 до 5

# Можно получить новый итератор и начать перебор заново
iterator1 = iter(sequence)
print(next(iterator1))  # 1
print(next(iterator1))  # 2

# Другой итератор начинает с начала
iterator2 = iter(sequence)
print(next(iterator2))  # 1

# Функция iter() создает итератор из итерируемого объекта
numbers = [1, 2, 3]
iterator = iter(numbers)
print(next(iterator))  # 1
print(next(iterator))  # 2
print(next(iterator))  # 3

# Функция next() получает следующий элемент итератора
# Можно указать значение по умолчанию, если итератор исчерпан
letters = iter("abc")
print(next(letters, "конец"))  # a
print(next(letters, "конец"))  # b
print(next(letters, "конец"))  # c
print(next(letters, "конец"))  # конец (не вызывает исключение)

# Функция-генератор для создания последовательности чисел
def count_up_to(limit):
    count = 0
    while count < limit:
        yield count
        count += 1

# Использование генератора
for i in count_up_to(5):
    print(i)  # Выведет числа от 0 до 4

# Генератор Фибоначчи
def fibonacci(n):
    a, b = 0, 1
    for _ in range(n):
        yield a
        a, b = b, a + b

# Выведем первые 10 чисел Фибоначчи
print(list(fibonacci(10)))  # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

# Списковое включение - создает весь список сразу
squares_list = [x**2 for x in range(10)]
print(squares_list)  # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

# Генераторное выражение - создает итератор
squares_generator = (x**2 for x in range(10))
print(squares_generator)  #  at 0x...>
print(list(squares_generator))  # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

# Обработка больших объемов данных
# Эта операция не загружает весь файл в память
def read_large_file(file_path):
    with open(file_path, 'r') as file:
        for line in file:
            yield line.strip()

# Эффективная обработка файла построчно
# for line in read_large_file('huge_file.txt'):
#     process_line(line)

def echo_generator():
    value = yield "Начало"
    print(f"Получено: {value}")
    
    value = yield "Продолжение"
    print(f"Получено: {value}")
    
    yield "Конец"

gen = echo_generator()
print(next(gen))     # "Начало"
print(gen.send("Привет"))  # Выводит "Получено: Привет" и возвращает "Продолжение"
print(gen.send("Мир"))     # Выводит "Получено: Мир" и возвращает "Конец"

def generator_with_exception():
    try:
        yield 1
        yield 2
        yield 3
    except ValueError:
        yield "Поймано исключение ValueError"
    yield 4

gen = generator_with_exception()
print(next(gen))  # 1
print(next(gen))  # 2
print(gen.throw(ValueError, "Ошибка"))  # "Поймано исключение ValueError"
print(next(gen))  # 4

def closeable_generator():
    try:
        yield 1
        yield 2
        yield 3
    except GeneratorExit:
        print("Генератор закрыт")
    
gen = closeable_generator()
print(next(gen))  # 1
gen.close()       # Выведет "Генератор закрыт"

try:
    print(next(gen))  # Вызовет StopIteration
except StopIteration:
    print("Генератор уже исчерпан")

# Этот цикл
for item in iterable:
    # обработка item
    
# Примерно эквивалентен этому коду
iterator = iter(iterable)
while True:
    try:
        item = next(iterator)
        # обработка item
    except StopIteration:
        break

import itertools

# Создание собственного бесконечного генератора
def count_forever(start=0, step=1):
    num = start
    while True:
        yield num
        num += step

# Используем с ограничением (иначе цикл будет бесконечным)
counter = count_forever(10, 5)
for _ in range(5):
    print(next(counter))  # 10, 15, 20, 25, 30

# Встроенные функции из itertools
# count - бесконечная арифметическая прогрессия
for i in itertools.islice(itertools.count(0, 2), 5):
    print(i)  # 0, 2, 4, 6, 8

# cycle - бесконечный цикл по итерируемому объекту
colors = itertools.cycle(['red', 'green', 'blue'])
for _ in range(5):
    print(next(colors))  # red, green, blue, red, green

# repeat - бесконечное повторение объекта
for i in itertools.islice(itertools.repeat("привет"), 3):
    print(i)  # привет, привет, привет

import itertools

# chain - объединяет несколько итераторов в один
combined = itertools.chain([1, 2, 3], ['a', 'b'], (True, False))
print(list(combined))  # [1, 2, 3, 'a', 'b', True, False]

# islice - выбирает подмножество из итератора
print(list(itertools.islice(range(10), 3, 8)))  # [3, 4, 5, 6, 7]

# zip_longest - аналог zip, но продолжает до самого длинного итератора
list1 = [1, 2, 3]
list2 = ['a', 'b', 'c', 'd', 'e']
print(list(itertools.zip_longest(list1, list2, fillvalue='?')))
# [(1, 'a'), (2, 'b'), (3, 'c'), ('?', 'd'), ('?', 'e')]

# product - декартово произведение итераторов
print(list(itertools.product('AB', '12')))
# [('A', '1'), ('A', '2'), ('B', '1'), ('B', '2')]

# combinations - все возможные комбинации длины r без повторений
print(list(itertools.combinations('ABC', 2)))
# [('A', 'B'), ('A', 'C'), ('B', 'C')]

# permutations - все возможные перестановки
print(list(itertools.permutations('ABC', 2)))
# [('A', 'B'), ('A', 'C'), ('B', 'A'), ('B', 'C'), ('C', 'A'), ('C', 'B')]

# Пример, демонстрирующий ленивые вычисления
def fibonacci_generator():
    a, b = 0, 1
    while True:
        yield a
        a, b = b, a + b

# Создаем генератор, но он не вычисляет значения сразу
fib = fibonacci_generator()

# Вычисляем только первые 10 чисел Фибоначчи
first_ten = list(itertools.islice(fib, 10))
print(first_ten)  # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]

import threading
import time

# Функция, которую будем выполнять в отдельном потоке
def print_numbers(name, delay):
    for i in range(5):
        time.sleep(delay)  # Имитация работы
        print(f"{name}: {i}")

# Создание потоков
thread1 = threading.Thread(target=print_numbers, args=("Thread 1", 0.5))
thread2 = threading.Thread(target=print_numbers, args=("Thread 2", 1))

# Запуск потоков
thread1.start()
thread2.start()

# Ожидание завершения потоков
thread1.join()
thread2.join()

print("Все потоки завершили работу")

# Вывод будет вперемешку, так как потоки выполняются параллельно:
# Thread 1: 0
# Thread 2: 0
# Thread 1: 1
# Thread 1: 2
# Thread 2: 1
# Thread 1: 3
# Thread 2: 2
# Thread 1: 4
# Thread 2: 3
# Thread 2: 4
# Все потоки завершили работу

import threading
import time

class MyThread(threading.Thread):
    def __init__(self, name, delay):
        super().__init__()
        self.name = name
        self.delay = delay
    
    def run(self):
        print(f"Поток {self.name} начал работу")
        for i in range(5):
            time.sleep(self.delay)
            print(f"{self.name}: {i}")
        print(f"Поток {self.name} завершил работу")

# Создание экземпляров потоков
thread1 = MyThread("Thread 1", 0.5)
thread2 = MyThread("Thread 2", 1)

# Запуск потоков
thread1.start()
thread2.start()

# Ожидание завершения потоков
thread1.join()
thread2.join()

print("Все потоки завершили работу")

import threading
import time

def worker():
    print(f"{threading.current_thread().name} начал работу")
    time.sleep(2)
    print(f"{threading.current_thread().name} закончил работу")

# Создаем поток
thread = threading.Thread(target=worker, name="WorkerThread")

# Проверяем, активен ли поток
print(f"Поток активен перед запуском: {thread.is_alive()}")

# Запускаем поток
thread.start()

# Проверяем, активен ли поток после запуска
print(f"Поток активен после запуска: {thread.is_alive()}")

# Ждем завершения потока
thread.join()

# Проверяем, активен ли поток после завершения
print(f"Поток активен после завершения: {thread.is_alive()}")

import threading
import time

def daemon_worker():
    while True:
        print("Демон работает")
        time.sleep(1)

# Создаем демон-поток
daemon_thread = threading.Thread(target=daemon_worker, daemon=True)
daemon_thread.start()

# Основной поток выполняет свою работу
time.sleep(3)

print("Основной поток завершил работу")
# Программа завершится здесь, даже если демон-поток еще работает

# Пример: GIL ограничивает производительность CPU-bound задач
import threading
import time

def cpu_bound_task(n):
    # Имитация CPU-интенсивной задачи
    count = 0
    for i in range(n):
        count += i
    return count

def single_thread():
    start = time.time()
    cpu_bound_task(100_000_000)
    cpu_bound_task(100_000_000)
    end = time.time()
    print(f"Однопоточное выполнение: {end - start:.2f} секунд")

def multi_thread():
    start = time.time()
    
    t1 = threading.Thread(target=cpu_bound_task, args=(100_000_000,))
    t2 = threading.Thread(target=cpu_bound_task, args=(100_000_000,))
    
    t1.start()
    t2.start()
    
    t1.join()
    t2.join()
    
    end = time.time()
    print(f"Многопоточное выполнение: {end - start:.2f} секунд")

# Из-за GIL, многопоточная версия работает примерно так же, 
# или даже медленнее, чем однопоточная
single_thread()
multi_thread()

import threading
import time

# Общий ресурс
counter = 0
counter_lock = threading.Lock()

def increment(n):
    global counter
    for _ in range(n):
        # Захват блокировки
        counter_lock.acquire()
        try:
            # Критическая секция
            counter += 1
        finally:
            # Освобождение блокировки
            counter_lock.release()

# Альтернативный синтаксис с использованием контекстного менеджера
def increment_with_context(n):
    global counter
    for _ in range(n):
        with counter_lock:  # Автоматически захватывает и освобождает блокировку
            counter += 1

# Создаем потоки
thread1 = threading.Thread(target=increment, args=(1000000,))
thread2 = threading.Thread(target=increment, args=(1000000,))

# Запускаем потоки
thread1.start()
thread2.start()

# Ожидаем завершения
thread1.join()
thread2.join()

print(f"Финальное значение счетчика: {counter}")  # 2000000 (верно)

import threading

rlock = threading.RLock()

def function_with_nested_locks():
    with rlock:
        print("Внешняя секция")
        
        # Тот же поток может снова захватить rlock
        with rlock:
            print("Вложенная секция")
        
        print("Обратно во внешней секции")

# Если бы мы использовали обычный Lock, произошла бы мертвая блокировка (deadlock)
thread = threading.Thread(target=function_with_nested_locks)
thread.start()
thread.join()

import threading
import time

# Реализация очереди с ограниченным размером
class BoundedQueue:
    def __init__(self, size):
        self.queue = []
        self.size = size
        self.condition = threading.Condition()
    
    def put(self, item):
        with self.condition:
            # Если очередь полна, ждем
            while len(self.queue) >= self.size:
                print("Очередь полна, ожидание...")
                self.condition.wait()
            
            # Добавляем элемент и уведомляем потоки, которые могут ждать
            self.queue.append(item)
            print(f"Добавлен элемент: {item}")
            self.condition.notify()
    
    def get(self):
        with self.condition:
            # Если очередь пуста, ждем
            while len(self.queue) == 0:
                print("Очередь пуста, ожидание...")
                self.condition.wait()
            
            # Извлекаем элемент и уведомляем потоки, которые могут ждать
            item = self.queue.pop(0)
            print(f"Извлечен элемент: {item}")
            self.condition.notify()
            return item

# Пример использования
queue = BoundedQueue(2)

def producer():
    for i in range(5):
        queue.put(i)
        time.sleep(0.5)

def consumer():
    for _ in range(5):
        item = queue.get()
        time.sleep(1)

# Запускаем потоки производителя и потребителя
prod_thread = threading.Thread(target=producer)
cons_thread = threading.Thread(target=consumer)

prod_thread.start()
cons_thread.start()

prod_thread.join()
cons_thread.join()

import threading
import time
import random

# Имитация пула соединений с базой данных
class ConnectionPool:
    def __init__(self, max_connections):
        self.semaphore = threading.Semaphore(max_connections)
    
    def get_connection(self):
        self.semaphore.acquire()
        return f"Connection-{random.randint(1, 1000)}"
    
    def release_connection(self, connection):
        print(f"Освобождение {connection}")
        self.semaphore.release()

# Создаем пул с максимум 3 соединениями
pool = ConnectionPool(3)

def worker(worker_id):
    connection = pool.get_connection()
    print(f"Работник {worker_id} получил {connection}")
    
    # Имитация работы с соединением
    time.sleep(random.uniform(1, 3))
    
    pool.release_connection(connection)
    print(f"Работник {worker_id} завершил работу")

# Запускаем 5 потоков, но одновременно будут работать максимум 3
threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

import threading
import time

# Создаем объект события
event = threading.Event()

def waiter(name):
    print(f"{name} ожидает событие")
    event.wait()  # Поток блокируется здесь до установки события
    print(f"{name} получил уведомление!")

def setter():
    print("Готовим данные...")
    time.sleep(2)
    print("Данные готовы, уведомляем ожидающие потоки")
    event.set()  # Устанавливаем событие, разблокируя все ожидающие потоки

# Запускаем ожидающие потоки
for i in range(3):
    threading.Thread(target=waiter, args=(f"Ожидающий-{i}",)).start()

# Запускаем поток, который установит событие
threading.Thread(target=setter).start()

# Основной поток продолжает работу
time.sleep(5)

# Сбрасываем событие
event.clear()
print("Событие сброшено")

import concurrent.futures
import time
import random

def task(task_id):
    print(f"Задача {task_id} начала выполнение")
    # Имитация работы
    sleep_time = random.uniform(0.5, 2)
    time.sleep(sleep_time)
    print(f"Задача {task_id} завершена за {sleep_time:.2f} секунд")
    return task_id, sleep_time

# Создаем пул потоков с 3 рабочими потоками
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    # Отправляем задачи на выполнение
    futures = [executor.submit(task, i) for i in range(10)]
    
    # Ждем результаты по мере выполнения задач
    for future in concurrent.futures.as_completed(futures):
        task_id, sleep_time = future.result()
        print(f"Получен результат задачи {task_id}: {sleep_time:.2f} секунд")
        
# При выходе из блока with все потоки завершатся

# Альтернативный способ с map
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    # Выполняем map задач на пул потоков
    for task_id, sleep_time in executor.map(task, range(10, 15)):
        print(f"Map: Задача {task_id} выполнялась {sleep_time:.2f} секунд")

import threading

# Общая переменная
counter = 0

def increment(n):
    global counter
    for _ in range(n):
        # Здесь может произойти состояние гонки
        # Операция counter += 1 на самом деле состоит из трех шагов:
        # 1. Прочитать текущее значение counter
        # 2. Увеличить значение на 1
        # 3. Записать новое значение обратно в counter
        counter += 1

# Запускаем два потока, которые инкрементируют счетчик
t1 = threading.Thread(target=increment, args=(100000,))
t2 = threading.Thread(target=increment, args=(100000,))

t1.start()
t2.start()

t1.join()
t2.join()

print(f"Ожидаемое значение: 200000")
print(f"Фактическое значение: {counter}")
# Скорее всего, counter будет меньше 200000 из-за состояния гонки

import threading
import time

# Создаем две блокировки
lock1 = threading.Lock()
lock2 = threading.Lock()

def thread1_function():
    print("Поток 1: Пытается захватить блокировку 1")
    lock1.acquire()
    print("Поток 1: Блокировка 1 захвачена")
    
    time.sleep(0.5)  # Небольшая задержка для гарантии возникновения deadlock
    
    print("Поток 1: Пытается захватить блокировку 2")
    lock2.acquire()
    print("Поток 1: Блокировка 2 захвачена")
    
    # Освобождаем блокировки
    lock2.release()
    lock1.release()

def thread2_function():
    print("Поток 2: Пытается захватить блокировку 2")
    lock2.acquire()
    print("Поток 2: Блокировка 2 захвачена")
    
    time.sleep(0.5)  # Небольшая задержка для гарантии возникновения deadlock
    
    print("Поток 2: Пытается захватить блокировку 1")
    lock1.acquire()  # Здесь произойдет deadlock
    print("Поток 2: Блокировка 1 захвачена")
    
    # Освобождаем блокировки
    lock1.release()
    lock2.release()

# Запускаем потоки
t1 = threading.Thread(target=thread1_function)
t2 = threading.Thread(target=thread2_function)

t1.start()
t2.start()

# Эта программа "зависнет" из-за deadlock
# Чтобы избежать deadlock, всегда захватывайте блокировки в одинаковом порядке во всех потоках

import threading
import time

# Создаем две блокировки
lock1 = threading.Lock()
lock2 = threading.Lock()

def safe_thread1_function():
    print("Безопасный поток 1: Пытается захватить блокировку 1")
    if lock1.acquire(timeout=1):  # Добавляем таймаут
        print("Безопасный поток 1: Блокировка 1 захвачена")
        
        time.sleep(0.5)
        
        print("Безопасный поток 1: Пытается захватить блокировку 2")
        if lock2.acquire(timeout=1):  # Добавляем таймаут
            print("Безопасный поток 1: Блокировка 2 захвачена")
            # Работа с защищенными ресурсами
            lock2.release()
        else:
            print("Безопасный поток 1: Не удалось захватить блокировку 2. Освобождаем ресурсы.")
        
        lock1.release()
    else:
        print("Безопасный поток 1: Не удалось захватить блокировку 1")

def safe_thread2_function():
    # Захватываем блокировки в том же порядке, что и в safe_thread1_function
    print("Безопасный поток 2: Пытается захватить блокировку 1")
    if lock1.acquire(timeout=1):
        print("Безопасный поток 2: Блокировка 1 захвачена")
        
        time.sleep(0.5)
        
        print("Безопасный поток 2: Пытается захватить блокировку 2")
        if lock2.acquire(timeout=1):
            print("Безопасный поток 2: Блокировка 2 захвачена")
            # Работа с защищенными ресурсами
            lock2.release()
        else:
            print("Безопасный поток 2: Не удалось захватить блокировку 2. Освобождаем ресурсы.")
        
        lock1.release()
    else:
        print("Безопасный поток 2: Не удалось захватить блокировку 1")

import threading
import multiprocessing
import time

def cpu_intensive_task(n):
    # Имитация CPU-интенсивной задачи
    result = 0
    for i in range(n):
        result += i * i
    return result

# Функция для тестирования многопоточности
def test_threading():
    start_time = time.time()
    
    threads = []
    for _ in range(4):
        t = threading.Thread(target=cpu_intensive_task, args=(10000000,))
        threads.append(t)
        t.start()
    
    for t in threads:
        t.join()
    
    end_time = time.time()
    print(f"Многопоточность: {end_time - start_time:.2f} секунд")

# Функция для тестирования многопроцессорности
def test_multiprocessing():
    start_time = time.time()
    
    processes = []
    for _ in range(4):
        p = multiprocessing.Process(target=cpu_intensive_task, args=(10000000,))
        processes.append(p)
        p.start()
    
    for p in processes:
        p.join()
    
    end_time = time.time()
    print(f"Многопроцессорность: {end_time - start_time:.2f} секунд")

# Тестируем оба подхода
if __name__ == "__main__":
    test_threading()
    test_multiprocessing()
    # Для CPU-интенсивных задач многопроцессорность работает быстрее

import asyncio

# Определяем асинхронную функцию (корутину)
async def say_hello(name, delay):
    # await может использоваться только внутри async функций
    await asyncio.sleep(delay)  # Неблокирующая пауза
    print(f"Привет, {name}!")
    return f"{name} поприветствован"

# Создаем и запускаем асинхронные задачи
async def main():
    # Запускаем несколько корутин параллельно
    results = await asyncio.gather(
        say_hello("Алиса", 1),
        say_hello("Боб", 2),
        say_hello("Чарли", 3)
    )
    print(f"Результаты: {results}")

# Запускаем асинхронную программу
asyncio.run(main())  # В Python 3.7+

# Вывод (с задержками):
# Привет, Алиса! (через 1 сек)
# Привет, Боб! (через 2 сек)
# Привет, Чарли! (через 3 сек)
# Результаты: ['Алиса поприветствован', 'Боб поприветствован', 'Чарли поприветствован']

import asyncio

async def fetch_data(url):
    print(f"Начинаем загрузку данных с {url}")
    # Имитация сетевого запроса
    await asyncio.sleep(2)
    print(f"Данные с {url} загружены")
    return f"Данные от {url}"

async def process_data(data):
    print(f"Начинаем обработку {data}")
    # Имитация обработки данных
    await asyncio.sleep(1)
    print(f"Обработка {data} завершена")
    return f"Обработанные {data}"

async def main():
    # Получаем данные асинхронно
    raw_data = await fetch_data("example.com/api")
    
    # Обрабатываем данные асинхронно
    processed_data = await process_data(raw_data)
    
    print(f"Итоговый результат: {processed_data}")

asyncio.run(main())

import asyncio
import time

async def do_work(name, seconds):
    print(f"{name} начинает работу")
    await asyncio.sleep(seconds)
    print(f"{name} закончил работу после {seconds} секунд")
    return f"Результат от {name}"

async def main():
    # Создаем задачи из корутин
    task1 = asyncio.create_task(do_work("Задача 1", 3))
    task2 = asyncio.create_task(do_work("Задача 2", 1))
    task3 = asyncio.create_task(do_work("Задача 3", 2))
    
    # Запускаем все задачи параллельно
    start_time = time.time()
    
    # Ждем завершения всех задач
    results = await asyncio.gather(task1, task2, task3)
    
    end_time = time.time()
    print(f"Все задачи завершены за {end_time - start_time:.2f} секунд")
    print(f"Результаты: {results}")

asyncio.run(main())

# Вывод:
# Задача 1 начинает работу
# Задача 2 начинает работу
# Задача 3 начинает работу
# Задача 2 закончил работу после 1 секунд
# Задача 3 закончил работу после 2 секунд
# Задача 1 закончил работу после 3 секунд
# Все задачи завершены за 3.00 секунд
# Результаты: ['Результат от Задача 1', 'Результат от Задача 2', 'Результат от Задача 3']

import asyncio

async def delayed_task(delay):
    await asyncio.sleep(delay)
    return f"Задача с задержкой {delay} с"

async def main():
    # asyncio.gather - запускает несколько корутин параллельно
    # и ждет результаты всех
    results = await asyncio.gather(
        delayed_task(1),
        delayed_task(2),
        delayed_task(3)
    )
    print(f"gather: {results}")  # Все результаты в порядке задач
    
    # asyncio.wait_for - выполняет корутину с таймаутом
    try:
        result = await asyncio.wait_for(delayed_task(5), timeout=2)
        print(f"wait_for: {result}")
    except asyncio.TimeoutError:
        print("wait_for: Превышено время ожидания!")
    
    # asyncio.as_completed - возвращает итератор завершенных задач
    # в порядке их завершения
    tasks = [
        asyncio.create_task(delayed_task(3)),
        asyncio.create_task(delayed_task(1)),
        asyncio.create_task(delayed_task(2))
    ]
    
    for coro in asyncio.as_completed(tasks):
        result = await coro
        print(f"as_completed: {result}")  # Результаты в порядке завершения

asyncio.run(main())

# Вывод:
# gather: ['Задача с задержкой 1 с', 'Задача с задержкой 2 с', 'Задача с задержкой 3 с']
# wait_for: Превышено время ожидания!
# as_completed: Задача с задержкой 1 с
# as_completed: Задача с задержкой 2 с
# as_completed: Задача с задержкой 3 с

import asyncio

class AsyncContextManager:
    async def __aenter__(self):
        print("Входим в контекст")
        await asyncio.sleep(1)
        return "контекстный объект"
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("Выходим из контекста")
        await asyncio.sleep(0.5)

async def main():
    async with AsyncContextManager() as context:
        print(f"Внутри контекста с объектом: {context}")
        await asyncio.sleep(1)
        print("Работаем внутри контекста")

asyncio.run(main())

# Вывод:
# Входим в контекст
# Внутри контекста с объектом: контекстный объект
# Работаем внутри контекста
# Выходим из контекста

import asyncio

class AsyncCounter:
    def __init__(self, limit):
        self.limit = limit
        self.counter = 0
    
    def __aiter__(self):
        return self
    
    async def __anext__(self):
        if self.counter < self.limit:
            await asyncio.sleep(0.5)  # Имитация асинхронной работы
            self.counter += 1
            return self.counter - 1
        else:
            raise StopAsyncIteration

async def main():
    # Используем асинхронный итератор
    async for i in AsyncCounter(5):
        print(f"Получено число: {i}")

asyncio.run(main())

# Вывод (с паузами в 0.5 секунды):
# Получено число: 0
# Получено число: 1
# Получено число: 2
# Получено число: 3
# Получено число: 4

import asyncio

async def async_range(start, stop):
    for i in range(start, stop):
        await asyncio.sleep(0.5)  # Имитация асинхронной работы
        yield i

async def main():
    # Используем асинхронный генератор
    async for i in async_range(5, 10):
        print(f"Получено число: {i}")

asyncio.run(main())

# Вывод (с паузами в 0.5 секунды):
# Получено число: 5
# Получено число: 6
# Получено число: 7
# Получено число: 8
# Получено число: 9

# Для использования этого примера требуется установить aiohttp:
# pip install aiohttp

import asyncio
import aiohttp
import time

async def fetch_url(session, url):
    print(f"Запрос к {url}")
    async with session.get(url) as response:
        data = await response.text()
        print(f"Получен ответ от {url}, размер: {len(data)} байт")
        return len(data)

async def main():
    urls = [
        "https://python.org",
        "https://www.google.com",
        "https://github.com",
        "https://stackoverflow.com",
        "https://www.wikipedia.org"
    ]
    
    start_time = time.time()
    
    # Создаем сессию для всех запросов
    async with aiohttp.ClientSession() as session:
        # Запускаем все запросы параллельно
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
    
    end_time = time.time()
    
    print(f"Все запросы выполнены за {end_time - start_time:.2f} секунд")
    print(f"Общий размер данных: {sum(results)} байт")

# asyncio.run(main())  # Раскомментируйте для запуска

# Для использования этого примера требуется установить aiofiles:
# pip install aiofiles

import asyncio
import aiofiles

async def read_file(filename):
    try:
        async with aiofiles.open(filename, "r") as file:
            content = await file.read()
            print(f"Прочитано {len(content)} байт из файла {filename}")
            return content
    except Exception as e:
        print(f"Ошибка при чтении {filename}: {e}")
        return ""

async def write_file(filename, content):
    async with aiofiles.open(filename, "w") as file:
        await file.write(content)
        print(f"Записано {len(content)} байт в файл {filename}")

async def main():
    # Чтение нескольких файлов параллельно
    contents = await asyncio.gather(
        read_file("file1.txt"),
        read_file("file2.txt"),
        read_file("file3.txt")
    )
    
    # Объединяем содержимое
    combined = "\n".join(contents)
    
    # Записываем в новый файл
    await write_file("combined.txt", combined)

# asyncio.run(main())  # Раскомментируйте для запуска

import time
import asyncio
import threading
import multiprocessing

# Функция для тестирования (будет адаптирована для каждого подхода)
def io_bound_task(sleep_time):
    # Имитация I/O-операции
    time.sleep(sleep_time)
    return f"Задача завершена через {sleep_time} сек"

async def async_io_bound_task(sleep_time):
    # Асинхронная версия той же задачи
    await asyncio.sleep(sleep_time)
    return f"Async задача завершена через {sleep_time} сек"

def threading_test():
    start_time = time.time()
    
    # Создаем потоки
    threads = []
    for i in range(1, 6):
        thread = threading.Thread(target=io_bound_task, args=(i,))
        threads.append(thread)
        thread.start()
    
    # Ожидаем завершения всех потоков
    for thread in threads:
        thread.join()
    
    end_time = time.time()
    print(f"Threading подход: {end_time - start_time:.2f} секунд")

def multiprocessing_test():
    start_time = time.time()
    
    # Создаем процессы
    processes = []
    for i in range(1, 6):
        process = multiprocessing.Process(target=io_bound_task, args=(i,))
        processes.append(process)
        process.start()
    
    # Ожидаем завершения всех процессов
    for process in processes:
        process.join()
    
    end_time = time.time()
    print(f"Multiprocessing подход: {end_time - start_time:.2f} секунд")

async def asyncio_test():
    start_time = time.time()
    
    # Создаем и запускаем асинхронные задачи
    tasks = [async_io_bound_task(i) for i in range(1, 6)]
    await asyncio.gather(*tasks)
    
    end_time = time.time()
    print(f"Asyncio подход: {end_time - start_time:.2f} секунд")

def main():
    # Тестирование threading
    print("Запуск threading теста...")
    threading_test()
    
    # Тестирование multiprocessing
    print("Запуск multiprocessing теста...")
    multiprocessing_test()
    
    # Тестирование asyncio
    print("Запуск asyncio теста...")
    asyncio.run(asyncio_test())

# if __name__ == "__main__":
#     main()

# Ожидаемый вывод:
# Запуск threading теста...
# Threading подход: ~5.01 секунд
# Запуск multiprocessing теста...
# Multiprocessing подход: ~5.05 секунд
# Запуск asyncio теста...
# Asyncio подход: ~5.00 секунд

import asyncio
import logging
import traceback

# Настройка логирования
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("asyncio")

async def buggy_task():
    try:
        # Имитируем ошибку
        await asyncio.sleep(1)
        raise ValueError("Специально вызванная ошибка")
    except Exception:
        logger.error("Произошла ошибка:")
        logger.error(traceback.format_exc())

async def main():
    # Включаем отладочный режим asyncio
    # asyncio.get_event_loop().set_debug(True)  # В Python < 3.7
    
    # Запускаем задачу
    task = asyncio.create_task(buggy_task())
    
    # Отслеживаем статус задачи
    while not task.done():
        logger.debug(f"Задача {task} всё ещё выполняется")
        await asyncio.sleep(0.5)
    
    try:
        # Получаем результат (если была ошибка, она будет повторно вызвана здесь)
        result = task.result()
    except Exception as e:
        logger.error(f"Задача завершилась с ошибкой: {e}")
    else:
        logger.info(f"Задача успешно завершена с результатом: {result}")

# asyncio.run(main(), debug=True)  # debug=True включает отладочный режим в Python 3.7+