Так исторически сложилось, что последние 5 лет своей продуктовой разработки я работаю с микросервисами вокруг брокеров сообщений (преимущественно RabbitMQ и Kafka).
И все это время меня не покидало чувство некой неудовлетворенности и неполноценности инструментария, который был мне доступен.
Приходя из мира HTTP фреймворков, ты чувствуешь себя как будто на костылях - ни тебе hotreload'а, который есть практически в любом wsgi-asgi сервере, хочешь тестировать - поднимай контейнеры окружения или мокай зависимости (особенно удобно в CI, ага), не забудь о реконнектах, логировании, трассировке и тд и тп.
И вот, таская от сервиса к сервису ворох всех этих проблем (и код который эти проблемы решает), до меня дошла гениальная идея: оформить весь однотипный код, общий для всех сервисов в единый пакет!
Так появился на свет фреймворк Propan.
Зачем вам использовать Propan
Фреймворк слеплен по образу и подобию FastAPI, но для работы с брокерами сообщений и с учетом лично моих болей, которые возникали при работе с этим HTTP-фреймворком. Propan в большей степени открыт для расширения и не диктует вам, как именно его использовать.
От FastAPI мы имеем:
Валидацию и приведение типов входящих сообщений с помощью Pydantic
Систему внедрения зависимостей
Максимально простой и понятный всем и каждому способ написания приложения
Особенности, которые выделяют Propan из ряда других фреймворков и нативных библиотек для работы с брокерами сообщений:
Независимость от брокеров - код, который вы пишете, не зависит от используемого брокера. Вы можете легко мигрировать с RabbitMQ на Kafka или Nats при росте нагрузки.
RPC поверх MQ - вам не нужно думать, как превратить Messaging в RPC. Если запрос ожидает ответ - фреймворк сделает все за вас.
Тестируемость - фреймворк позволяет эмулировать поведение брокеров и тестировать ваши сервисы без необходимости подключения к внешним зависимостям.
Собственный CLI - позволяет управлять настройками приложения, количеством запущенных инстансов, генерирует шаблоны проекта, а также перезагружает ваш проект при изменениях в коде (мне этого очень не хаватало при локальной разработке).
На текущий момент фрейморк поддерживает работу с RabbitMQ, Redis Pub/Sub, Nats. Поддержка Kafka ожидается в ближайший месяц, затем - работа над генерацией схемы в соответсвии с AsyncAPI.
Пример
Давайте немного опустим все эти сухие слова и перейдем к практическому примеру.
Например, вот код "приложения" для работы с RabbitMQ с использование aio-pika
import asyncio
import aio_pika
async def main():
connection = await aio_pika.connect_robust("amqp://guest:guest@127.0.0.1/")
async with connection:
channel = await connection.channel()
queue = await channel.declare_queue("test_queue")
async with queue.iterator() as queue_iter:
async for message in queue_iter:
async with message.process():
print(message.body)
asyncio.run(main())
А вот, тот же пример с использование Propan
from propan import PropanApp, RabbitBroker
broker = RabbitBroker("amqp://guest:guest@localhost:5672/")
app = PropanApp(broker)
@broker.handle("test_queue")
async def base_handler(body: str): # тело сообщения будет приведено к `str`
print(body)
В нем вам не нужно самостоятельно объявлять все подключения, очереди, заниматься обработкой сообщения: вы просто пишете код, а фреймворк делает все за вас. Однако, при необходимости, у вас остается возможность управлять этим поведением вручную.
При этом, пример с использованием Propan дает вам все преимущества, описанные в предыдущем разделе.
Пример с использованием Redis или Nats идентичен с точностью до аргументов функций.
from propan import PropanApp
from propan import RabbitBroker
# from propan import RedisBroker
# from propan import NatsBroker
broker = RabbitBroker("amqp://guest:guest@localhost:5672/")
# broker = NatsBroker("nats://localhost:4222")
# broker = RedisBroker("redis://localhost:6379")
app = PropanApp(broker)
@broker.handle("test")
async def base_handler(body: str):
print(body)
Использование в HTTP сервисах
Очень часто нам нужно использовать логику работы с брокерами сообщений в рамках приложений, которые обрабатывает и HTTP-запросы.
Без проблем! Просто запустите и остановите Propan вместе с вашим приложением.
Пример с aiohttp
from aiohttp import web
from propan import RabbitBroker
broker = RabbitBroker("amqp://guest:guest@localhost:5672/")
@broker.handle("test")
async def base_handler(body):
print(body)
async def start_broker(app):
await broker.start()
async def stop_broker(app):
await broker.close()
app = web.Application()
app.on_startup.append(start_broker) # запускаем Propan при старте
app.on_cleanup.append(stop_broker) # и останавливаем вместе с приложением
web.run_app(app)
Кроме этого, если вы используете FastAPI, вы можете использовать Propan напрямую - как часть вашего приложения FastAPI.
from fastapi import FastAPI
from pydantic import BaseModel
from propan.fastapi import RabbitRouter
app = FastAPI()
router = RabbitRouter("amqp://guest:guest@localhost:5672")
class Incoming(BaseModel):
...
@router.event("test")
async def hello(body: Incoming):
print(body)
app.include_router(router)
Заключение
Не вижу смысла пересказывать всю документацию фреймворка, вы можете найти ее здесь. Если по ходу чтения статьи у вас возникли какие-то вопросы, возможно, ответы вы также найдете там.
Сейчас фреймворк активно развивается только моими силами. Если вы заинтересованы в его дальнейшем развитии, я буду крайне благодарен за любую помощь: указание на неточности/непонятные моменты в документации, написание кода, тестирование.
В любом случае, я буду рад любому фидбеку: критике, предложениям для реализации. Если вы решите использовать фреймворк в своих проектах - буду рад услышать о вашем опыте.
