📱Pyper.

Pyper — это гибкий инструмент для конкурентной и параллельной обработки данных, основанный на принципах функционального программирования.

🔹Применяется для:

🔘ETL-систем,
🔘Микросервисов для работы с данными,
🔘Сбора данных.

🔻Основные преимущества:

🔘Интуитивно понятный API: Простое и логичное использование. Единая абстракция для потоков, процессов и асинхронных задач.

🔘Функциональный подход: Построение конвейеров данных из функций Python. Код становится чистым и переиспользуемым.

🔘Безопасность: Автоматическая обработка задач, очистка ресурсов и устранение ошибок потоков. Забудьте о гонках данных и утечках памяти.

🔘Эффективность: Ленивое выполнение с использованием очередей, воркеров и генераторов.

🔘Чистый Python: Лёгкий фреймворк без сторонних зависимостей.

⚙️Пример использования:

import asyncio
import time
from pyper import task

def get_data(limit: int):
    for i in range(limit):
        yield i

async def step1(data: int):
    await asyncio.sleep(1)
    print("Async завершён:", data)
    return data

def step2(data: int):
    time.sleep(1)
    print("Sync завершён:", data)
    return data

def step3(data: int):
    for i in range(10_000_000):
        _ = i * i
    print("Тяжёлая операция завершена:", data)
    return data

async def main():
    pipeline = task(get_data, branch=True) \
        | task(step1, workers=20) \
        | task(step2, workers=20) \
        | task(step3, workers=20, multiprocess=True)

    total = 0
    async for output in pipeline(limit=20):
        total += output
    print("Итог:", total)

if __name__ == "__main__":
    asyncio.run(main())

❓Как это работает?

Конвейер: задачи связываются с помощью оператора | (аналог Unix pipe).

Виды задач:

✔️Асинхронные задачи для IO (step1).
✔️Потоки для синхронных IO-задач (step2).
✔️Процессы для CPU-нагруженных операций (step3).

Данные передаются между задачами через очереди для обеспечения конкурентного выполнения.

✅Попробуйте Pyper, чтобы ускорить обработку данных!

#Python #Pyper #DataProcessing #Concurrency #Parallelism #ETL #OpenSource #FunctionalProgramming

📱Ссылка на репозиторий:
https://github.com/pyper-dev/pyper

📱Bocpy.

Bocpy — это библиотека для Python, реализующая принципиально новую парадигму параллельного программирования — поведенческо-ориентированную конкурентность (Behavior-Oriented Concurrency, BOC).

В параллельном коде на Python часто возникают состояния гонки (data races), а с появлением сборки без GIL (free-threaded Python) защищать данные становится еще сложнее. bocpy решает эту проблему фундаментально: данные разделяются так, что каждый поток выполнения («поведение») получает уникальное временное владение объектом. Больше не нужно вручную управлять блокировками (lock.acquire()), а риск дедлоков исключен на уровне архитектуры.

🔹Главные фичи и преимущества:

🔘Фокус на потоке данных: Задача разработчика смещается с решения проблем параллельного доступа к простой организации логики. Функции оборачиваются в декораторы @when и со стороны выглядят как обычный линейный код.

🔘Настоящая многоядерность: Библиотека написана на C и использует механизм суб-интерпретаторов Python. Начиная с Python 3.12, где у каждого суб-интерпретатора свой собственный GIL (PEP 684), bocpy обеспечивает честный параллелизм и показывает практически линейное масштабирование (до 7.5× ускорения на 8 воркерах).

🔘Безопасность по умолчанию: Под капотом движок на C реализует двухфазное запирание (2PL) и MCS-очереди. Потоки изолированы друг от друга, поэтому данные обрабатываются так, будто программа работает в один поток.

🔘Поддержка тяжелых объектов (PinnedCown): Если объект не может быть передан между интерпретаторами (например, контекст GPU, файловые дескрипторы или UI-виджеты), его можно закрепить за главным потоком, не ломая параллельную работу остальных воркеров.

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

Вместо сложной синхронизации вы объявляете разделяемые ресурсы как Cown (concurrently-owned objects) и указываете, какие функции должны дождаться их готовности:

from bocpy import Cown, when, wait

# Создаем защищенные ресурсы
cheese = Cown(None)
meal = Cown(None)

# Задачи запустятся параллельно на разных ядрах
@when(cheese)
def buy_cheese(cheese):
    cheese.value = "Пармезан"

@when(meal)
def order_meal(meal):
    meal.value = "Салат"

# Этот блок выполнится автоматически, как только ОБОИ ресурсы освободятся
@when(cheese, meal)
def eat_dinner(cheese, meal):
    print(f"Ужинаем: {meal.value} и {cheese.value}!")

wait() # Ждем завершения всех задач

📈Поддержка версий Python:

🔘Python 3.10 / 3.11: Полная совместимость, но из-за общего глобального GIL задачи будут выполняться последовательно. Подходит для отладки и портативности.

🔘Python 3.12+: Идеальный вариант. Задачи распределяются по суб-интерпретаторам со своими GIL и работают на полную мощность процессора.

🔘Python 3.13t+ (Free-threaded): Полная нативная поддержка сборки без GIL.

🔧Установка:

pip install bocpy

#Python #bocpy #Concurrency #Parallelism #GIL #Multithreading #SubInterpreters #Backend #OpenSource

📱Ссылка на репозиторий:
https://github.com/microsoft/bocpy