Если у вас есть интересные опенсорсные проекты, про которые вы хотите рассказать, то пишите в чат.
С вас пост. С меня редактура и размещение. Давайте помогать друг другу!
А сегодняшний пост будет про очень прикольную библиотеку https://github.com/airtai/faststream от ее автора.
FastStream
Это современный фреймворк для разработки асинхронных сервисов поверх брокеров сообщений. Он взлетел за счет очень простого, интуитивного API:
from faststream import FastStream
from faststream.rabbit import RabbitBroker
broker = RabbitBroker()
app = FastStream(broker)
@broker.subscriber("in-queue")
@broker.publisher("out-queue")
async def handle_msg(user: str) -> str:
return f"User: {user} registered"
Но за простотой кроется достаточно интересное внутреннее устройство. Основная трудность, с которой борется FastStream (и почему у инструмента нет аналогов) - двухэтапная инициализация объектов. Это значит, что все вложенные объекты и данные, необходимые для функционирования "запчастей" не известны на момент их создания через
__init__ и должны быть доставлены позже.Как, например, в случае с мидлварями и декомпозицией приложения на отдельные router'ы
from faststream.nats import NatsBroker, NatsRouter
from faststream.nats.prometheus import NatsPrometheusMiddleware
router = NatsRouter()
publisher = router.publisher("out")
@router.subscriber("in")
async def handler(msg):
await publisher.publish("in")
broker = NatsBroker(middlewares=[NatsPrometheusMiddleware()])
broker.include_router(router)
На момент регистрации ни
publisher, ни subscriber ничего не знают о своих будущих мидлварях. Они создаются позже, в брокере. Что значит: на момент включения router'а в брокер мы должны передать подобные зависимости в роутер. FastAPI, например, решает эту проблему путем создания новых эндпоинтов как копий из экземпляра router'a. Однако, тут данный подход не сработает - тот же publisher используется внутри кода обработчика. Пересоздавать объекты мы не можем - старая ссылка должна быть валидна.Все объекты должны создаваться готовыми к использования настолько, насколько возможно, но также поддерживать операции переноса между различными контейнерами (роутеры / брокеры).
Более того, мидлвари и другие объекты, передаваемые в
Broker.__init__ - самая безобидная часть айсберга. Большая часть объектов требует наличия реального объекта connection к брокеру (который появляется только после асинхронного await broker.start() ). Взглянем на небольшой кусочек кода aiokafka (используется внутри FastStream):
consumer = AIOKafkaConsumer(...)
await consumer.start()
async for msg in consumer:
...
Как мы видим, для чтения сообщений необходим объект Consumer'а (который и держит
connection ). Соответственно, нам нобходимо доставить этот объект до subscriber'ов FastStream уже после запуска приложения.При запуске брокера, мы должны рекурсивно пройтись по всему дереву вложенных объектов и проинициализаровать его повторно реальными объектами, необходимыми для функционирования компонентов. Все это развестистое дерево двухэтапной инициализации с необходимостью сохранения валидности всех ссылок на уже созданные объекты приводит к довольно сложной, но интересной внутренней структуре проекта.
И это только небольшая часть сложностей, с которой вынужден бороться фреймворк! А там еще есть:
- in-memory тестирование
- собственный DI вдохновленный FastAPI
- сериализация на интроспекции типов
- поддержка разных бекендов: Kafka, RabbitMQ, Redis, NATS
- свой CLI
- много всякого-разного!
Если вы ищете интересный проект для участия в Open Source - FastStream сейчас нуждается в контрибуторах: ревью PR'ов, участие в обсуждениях, большие и маленькие фичи, правки в документацию - мы будем рады любому участию!
* Telegram группа проекта
* Доклад от создателя фреймворка с PiterPy