Structured Concurrency в Python 3.11+ (TaskGroup vs Gather)

Если вы до сих пор используете asyncio.gather() для запуска конкурентных задач, вы, вероятно, теряете контроль над ошибками.

Главная проблема gather: если одна таска падает, остальные продолжают работать (если не стоит return_exceptions=False, который убивает всё сразу, но "грязно"). Плюс, отлавливать несколько ошибок одновременно через try / except - это боль.

В Python 3.11+ (и через trio / anyio раньше) стандартом стала Structured Concurrency через asyncio.TaskGroup.

Как это выглядит:

import asyncio

async def task_fail(name, sec):
    await asyncio.sleep(sec)
    raise ValueError(f"Error in {name}")

async def task_ok(name, sec):
    await asyncio.sleep(sec)
    print(f"Task {name} done")

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(task_ok("A", 1))
            tg.create_task(task_fail("B", 2)) # Упадет через 2 сек
            tg.create_task(task_fail("C", 1.5)) # Упадет через 1.5 сек
            
    except* ValueError as eg: 
        # Обратите внимание на except* (ExceptionGroup)
        print(f"Caught errors: {eg.exceptions}")

# Запуск
# asyncio.run(main())

Почему TaskGroup это выбор Middle+:

1. Атомарность скоупа: Если одна задача внутри контекстного менеджера падает, TaskGroup автоматически отменяет (cancel) все остальные запущенные задачи в группе. Вы не оставляете "зомби-процессы" в фоне.
2. Exception Groups: Если упало несколько задач одновременно (или во время отмены), Python соберет их в ExceptionGroup. Конструкция except* позволяет элегантно обрабатывать дерево исключений.

Вывод:
Для скриптов gather сойдет. Для надежных сервисов, где важна консистентность состояния и корректный шатдаун корутин - переезжайте на TaskGroup.

#asyncio #python311 #bestpractices #concurrency

📲 Мы в MAX

👉@BookPython

Списковые включения (list comprehensions) могут содержать несколько операторов for и if:

In : [(x, y) for x in range(3) for y in range(3)]
Out: [
    (0, 0), (0, 1), (0, 2),
    (1, 0), (1, 1), (1, 2),
    (2, 0), (2, 1), (2, 2)
]

Можно также добавлять условия if для фильтрации значений:

In : [
    (x, y)
    for x in range(3)
    for y in range(3)
    if x != 0
    if y != 0
]
Out: [(1, 1), (1, 2), (2, 1), (2, 2)]

Любое выражение с for и if может использовать все переменные, определённые ранее:

In : [
    (x, y)
    for x in range(3)
    for y in range(x + 2)
    if x != y
]
Out: [
    (0, 1),
    (1, 0), (1, 2),
    (2, 0), (2, 1), (2, 3)
]

Вы можете комбинировать for и if в любом порядке:

In : [
    (x, y)
    for x in range(5)
    if x % 2
    for y in range(x + 2)
    if x != y
]
Out: [
    (1, 0), (1, 2),
    (3, 0), (3, 1), (3, 2), (3, 4)
]

📲 Мы в MAX

👉@BookPython

🐍 Как улучшить читаемость кода в Python?

Читаемый код - это не роскошь, а необходимость. Если ваш код трудно понять, даже если он работает, это плохой код. Сегодня разберем несколько простых, но мощных приемов, которые сделают ваш код более понятным.

1️⃣ Используйте говорящие имена переменных
Плохой пример:

a = 10
b = 20
c = a + b

Хороший пример:

price = 10
tax = 20
total_cost = price + tax

Теперь сразу понятно, что делает код!

2️⃣ Разбивайте код на функции
Вместо длинных кусков кода, используйте функции:

def calculate_total(price, tax):
    return price + tax

total_cost = calculate_total(10, 20)

Теперь код можно переиспользовать и проще тестировать.

3️⃣ Следуйте PEP 8
Форматирование кода влияет на его читаемость. Например, пробелы вокруг операторов делают код более понятным:

# Плохо
total=price+tax
# Хорошо
total = price + tax

Пользуйтесь black или flake8, чтобы следить за стилем.

4️⃣ Избегайте магических чисел
Если в коде встречаются непонятные числа, лучше заменить их на константы:

# Плохо
if age > 18:
    print("Взрослый")

# Хорошо
LEGAL_AGE = 18
if age > LEGAL_AGE:
    print("Взрослый")

5️⃣ Используйте list comprehensions
Вместо:

numbers = [1, 2, 3, 4, 5]
squared_numbers = []
for num in numbers:
    squared_numbers.append(num ** 2)

Лучше:

squared_numbers = [num ** 2 for num in numbers]

Чище и лаконичнее!

Читаемый код делает разработку приятнее, ускоряет исправление багов и упрощает поддержку. Напишите в комментариях, какие еще приемы вы используете для улучшения читаемости кода! 👇

📲 Мы в MAX

👉@BookPython

📌 Декораторы в Python: как они работают и зачем нужны?

Сегодня я покажу вам, как работают декораторы в Python и зачем они вообще нужны.

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

Допустим, у нас есть функция, которая просто выводит «Hello, world!»:

def greet():
    print("Hello, world!")

Теперь мы хотим, чтобы перед выполнением этой функции выполнялся какой-то код, например, логирование. Вместо изменения greet(), мы создадим декоратор:

def log_decorator(func):
    def wrapper():
        print(f"Вызов функции {func.__name__}")
        return func()
    return wrapper

И теперь используем его:

@log_decorator
def greet():
    print("Hello, world!")

greet()

👉 Вывод:

Вызов функции greet
Hello, world!

Как это работает?
1. Декоратор принимает функцию (func) в качестве аргумента.
2. Внутри создаётся вложенная функция wrapper(), которая выполняет дополнительную логику перед вызовом func().
3. wrapper() возвращается вместо func, фактически подменяя её.

Можно даже передавать аргументы в декорируемую функцию:

def log_decorator(func):
    def wrapper(*args, **kwargs):
        print(f"Вызов {func.__name__} с аргументами: {args}, {kwargs}")
        return func(*args, **kwargs)
    return wrapper

@log_decorator
def add(a, b):
    return a + b

print(add(3, 5))

👉 Вывод:

Вызов add с аргументами: (3, 5), {}
8

🔥 Декораторы это мощный инструмент, который делает код чище и удобнее. Если ещё не использовали их в проектах, самое время попробовать!

📲 Мы в MAX

👉@BookPython

Профилируем Python в продакшене: почему cProfile не подходит, и чем хорош py-spy

Когда на проде начинает течь память или скачет CPU, первая мысль, подключить профайлер. Стандартный cProfile, это детерминированный профайлер. Он хукает каждый вызов функции.

📉 Цена: Оверхед может замедлить приложение в 2-5 раз. На нагруженном проде это означает добить сервис окончательно.

Для Live-систем нужен Sampling Profiler (сэмплирующий профайлер). Золотой стандарт сейчас - py-spy.

Как это работает?
py-spy написан на Rust. Он работает как внешний процесс, который читает память вашего Python-процесса (через системные вызовы, аналогично gdb). Он делает «снимки» стека вызовов с высокой частотой (по дефолту 100 раз в секунду).

✅ Результат: Оверхед стремится к нулю. Код инструментализировать не нужно. Рестарт сервиса не нужен.

Два главных режима работы:

1. Live View (как htop, но для функций)
Посмотреть в реальном времени, в каких функциях процесс проводит больше всего времени.

# Нужно только знать PID процесса
py-spy top --pid 12345

Вы увидите список функций, отсортированный по OwnTime (время внутри функции) и TotalTime (время с учетом дочерних вызовов).

2. Flame Graph (Огненный граф)
Для глубокого анализа лучше записать работу сервиса за период и визуализировать стек.

py-spy record -o profile.svg --pid 12345 --duration 60

Вы получите SVG-файл. Чем шире полоска, тем больше времени занимает функция. Вертикаль - это глубина стека. Сразу видно, кто «съел» процессорное время.

Нюансы для Middle+:

- GIL: py-spy умеет показывать, держит ли функция GIL. Добавьте флаг --gil.
- Docker/K8s: Так как py-spy использует системный вызов ptrace, контейнеру нужны привилегии. В Kubernetes часто нужно добавить securityContext: capabilities: add: ["SYS_PTRACE"] подам, чтобы иметь возможность профилировать их на лету.


#profiling #optimization #pyspy #debugging #python

📲 Мы в MAX

👉@BookPython

Ловушка замыканий: Почему ваши лямбды в цикле сломаны (Late Binding)

Вы пишете код, который генерирует список функций (например, колбэки для кнопок в UI или динамические фильтры). Кажется, что все логично, но на выходе получаете сюрприз.

Проблемный код:

# Хотим создать 3 функции, которые возвращают 0, 1 и 2 соответственно
funcs = []
for i in range(3):
    funcs.append(lambda: i)

# Проверяем
results = [f() for f in funcs]
print(results) 
# Ожидание: [0, 1, 2]
# Реальность: [2, 2, 2]

Почему так происходит?
Это называется Late Binding (позднее связывание).
В Python замыкания (closures) захватывают переменные по ссылке, а не по значению.
Когда вы объявляете lambda: i, Python не сохраняет текущее число 0, 1 или 2. Он сохраняет инструкцию: «когда меня вызовут, пойди в локальную область видимости, найди переменную с именем i и возьми ее значение».

К моменту, когда вы начинаете вызывать функции из списка results, цикл for уже завершился. Переменная i в этой области видимости навсегда осталась равной 2. Все три лямбды смотрят на одну и ту же переменную i.

Как лечить?

Есть два каноничных способа заставить Python запомнить значение «здесь и сейчас».

1. Аргумент по умолчанию (Hack way)
Значения аргументов по умолчанию вычисляются в момент определения функции.

funcs = []
for i in range(3):
    # i=i создает локальную переменную i внутри функции 
    # и присваивает ей текущее значение i из цикла
    funcs.append(lambda i=i: i) 

Это работает быстро, но выглядит немного грязно и может сбить с толку линтеры или коллег.

2. functools.partial (Enterprise way)
Более чистый и явный способ. partial создает новый callable-объект, «замораживая» переданные аргументы.

from functools import partial

funcs = []
for i in range(3):
    # Здесь значение i фиксируется жестко
    funcs.append(partial(lambda x: x, i))

Где это стреляет в реальной жизни?

- Генерация command для кнопок в Tkinter/PyQt.
- Динамическое создание task в asyncio циклах.
- Патчинг тестов в циклах.

Не дайте переменным пережить свое время.

#python #internals #functionalprogramming #gotchas

📲 Мы в MAX

👉@BookPython

⚡ Как ускорить код на Python? Используем map, filter, reduce ⚡

Привет, друзья! Сегодня расскажу о том, как можно ускорить выполнение кода, заменяя обычные циклы на встроенные функции map(), filter() и reduce(). Эти инструменты позволяют писать более компактный, читаемый и быстрый код.

✅ map()
Функция map() применяется к каждому элементу последовательности и возвращает новый итератор.

❌ Обычный способ:

numbers = [1, 2, 3, 4, 5]
squared = []
for num in numbers:
    squared.append(num ** 2)

✅ Быстрее с map():

numbers = [1, 2, 3, 4, 5]
squared = list(map(lambda x: x ** 2, numbers))

За счёт того, что map() использует C-оптимизированную логику, код выполняется быстрее.

✅ filter()
Фильтрует элементы последовательности по заданному условию.

❌ Медленный вариант:

numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
evens = []
for num in numbers:
    if num % 2 == 0:
        evens.append(num)

✅ Быстрее с filter():

numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
evens = list(filter(lambda x: x % 2 == 0, numbers))

Такой код читается легче и работает быстрее.

✅ reduce()
Позволяет выполнять кумулятивные операции (например, суммирование, умножение).

❌ Классический способ:

numbers = [1, 2, 3, 4, 5]
product = 1
for num in numbers:
    product *= num

✅ Быстрее с reduce():

from functools import reduce

numbers = [1, 2, 3, 4, 5]
product = reduce(lambda x, y: x * y, numbers)

Этот метод полезен, если нужно свести список к одному значению.

💡 Важно: reduce() чаще заменяют sum() или math.prod(), но для сложных операций он остаётся полезным.


Такие функции помогают писать код, который не только быстрее работает, но и легче читается.

📲 Мы в MAX

👉@BookPython