Mela: асинхронный фреймворк на Python для сервисов, работающих с RabbitMQ

WARNING: длинная вступительная часть. Если хотите перейти сразу к делу — листайте до Getting Started.

Вступление

В 2023 году писать сервисы, взаимодействующие друг с другом через RabbitMQ, всё ещё неоправданно сложно. Ещё больше сложностей возникает с тестированием бизнес-логики в них, с согласованием контрактов между ними, с организацией монорепозиториев.

В недрах компании Alem Research, ещё года с 19-го, я начал писать легковесный фреймворк, который по задумке должен был стать чем-то вроде Flask или FastAPI для работы с RabbitMQ.

Удалось ли? Пожалуй, да. При помощи этого фреймворка наши дата-саентисты смогли самостоятельно писать и упаковывать в докер сервисы, которые лёгким движением руки и парой перебиндов встраиваются в трубу.

Впрочем, первая версия работала ещё с kombu, а позже ядро было переписано на aio-pika, и фреймворк стал ещё и асинхронным. Что даёт асинхронность? Например, можно написать фетчер HTTP-страниц, который при увеличении prefetch_count будет фетчить параллельно огромное количество страниц, и даже сможет утилизировать сеть практически полностью, чего без асинхронности было бы довольно сложно добиться.

Кроме того, асинхронность позволяет легче лёгкого реализовать паттерн RPC через RabbitMQ в рамках одного процесса, что довольно удобно для тестирования. Да и интеграция с FastAPI становится очень удобной. А скорость обработки растёт в десятки раз.

В 2020-м мы совместно с фаундером и тогдашним генеральным директором решили, что фреймворк смело можно выкладывать в open-source. И в целом, казалось бы, всё хорошо.

Однако, есть и проблемы. Главная из них — это нехватка живых пользователей и контрибьюторов. Я ушёл из Alem Research, и в компании продолжать работать над фреймворком в данный момент некому. Иногда кто-то из компании пишет мне запросы по фреймворку, но мне очень лень выкатывать обновления для одной-единственной компании: это получается своего рода работа на работодателя, который тебе больше не платит 🙂

Я давно уже собирался написать питчинг-статью про Mela, чтобы привлечь новых пользователей, а возможно даже и контрибьюторов, но не решался, потому что код был очень далёк от идеала. Да-да, меня просто жрал внутренний перфекционизм и стыд за то, чтобы выкатить что-то не очень красивое. Но сегодня я понял, что стыдиться мне особо нечего: да, архитектура ядра не идеальна, но за последние пару месяцев, пока я сидел без работы, я уже продумал её со всех сторон, и знаю, что делать дальше. В общем, красивая версия не готова, но у меня есть план.

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

Извините за долгое вступление. Мы наконец-то отправляемся в путь.

Getting started

Конечно же, всё начинается с:

pip install mela=1.1.1

Версию я указал чтобы статья оставалась релевантной даже после обновлений.

Давайте напишем сервис, который будет получать сообщение из очереди, выводить его body распарсенный в json, и возвращать сообщение обратно.

# app.py  from mela import Mela  app = Mela(__name__)   @app.service("printer") def printer(body, message):     print(body)     return body   if __name__ == '__main__':     app.run() 

Как вы можете догадаться, это ещё не всё, что требуется для запуска приложения: здесь нет никакой информации об очередях и подключении. Это потому что она хранится в файле application.yml.

# application.yml connections:   default:     host: localhost     port: 5672     username: user     password: bitnami  services:   printer:     consumer:       exchange: general-sentiment-x       routing_key: general-sentiment-q       queue: general-sentiment-q     publisher:       exchange: general-sentiment-x       routing_key: general-sentiment-q

Вот и всё.

Этот пример знакомит нас сразу с тремя самыми основными высокоуровневыми концепциями Мелы: Publisher, Consumer и Service.

Если с паблишером и консьюмером всё понятно, то про Service, пожалуй, поясню, что это элемент трубы, который состоит из комбинации консьюмера и паблишера: он консьюмит сообщения из определённой очереди, обрабатывает его, и паблишит в указанный эксчейндж. По сути, сервис состоит из паблишера и консьюмера, что в том числе отражено в Yaml-файле.

Функция, которая находится под декоратором @app.service(...), как не сложно догадаться, является коллбэком консьюмера, а то, что она возвращает — будет отправлено в паблишер, привязанный к сервису.

Pydantic

Какой современный фреймворк без Pydantic, правда? И их есть у меня.

# app.py from pydantic import BaseModel from datetime import datetime  from mela import Mela   app = Mela(__name__)   class Document(BaseModel):     text: str     url: str     date: datetime   @app.service('validator') def validator(body: Document) -> Document:     if '#' in body.url:         body.url = body.url.split('#')[0]     return body   if __name__ == '__main__':     app.run()

В данном случае мы используем new-style сигнатуру обработчика, и явно указывать второй аргумент, в который в предыдущем примере прилетел бы объект Message, нам не нужно.

body сначала переводится в json, потом этот json скармливается в класс Document.

При ошибках валидации ошибки будут выведены в консоль, а сообщение вернётся обратно в очередь. Если, конечно, в консьюмереrequeue_broken_messages=True (по умолчанию так) и если у консьюмера не задан dead letter exchange. То же самое будет происходить если в коллбэке зарейзится любая другая ошибка.

Управление ack/nack

Есть несколько способов управлять ответами. Давайте рассмотрим все сразу.

from datetime import datetime from datetime import timedelta from pydantic import BaseModel  from mela import IncomingMessage from mela import Mela from mela.components.exceptions import NackMessageError  app = Mela(__name__)   class Document(BaseModel):     text: str     url: str     date: datetime   @app.service("filter") async def filter_(body: Document, message: IncomingMessage):     if body.date > datetime.utcnow():         # First way: we can raise special exception with some `requeue` value         raise NackMessageError("We are not working with time travellers", requeue=False)     elif body.date < datetime.utcnow() - timedelta(days=365):         # Second way: we can manually nack message via IncomingMessage object         # As you can see, in this case we can't write any message about requeue reason.         # But it is still useful if you need to silently send message to DLX         await message.nack(requeue=False)  # Go to archive, dude      if body.url == '':         # Third way: we can raise almost any exception. The message should be or should not          # be requeued based on `requeue_broken_messages` value         raise AssertionError("Message without url is not acceptable")      return body   if __name__ == '__main__':     app.run() 

Чистый консьюмер

Консьюмер создаётся точно так же, как сервис, но с другим декоратором.

from pydantic import BaseModel from pydantic import EmailStr from mela import Mela  app = Mela(__name__)   class EmailNotification(BaseModel):     template_name: str     vars: dict     receiver: EmailStr   @app.consumer("email-sender") def printer(body: EmailNotification):     # Some Jinja2 and SMTP integration     pass   if __name__ == '__main__':     app.run() 

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

Ещё важно показать application.yml для консьюмера:

connections:   default:     host: localhost     port: 5672     username: user     password: bitnami  consumers:   email-sender:     exchange: notifications-x     exchange_type: topic     routing_key: "email.#"     queue: email-sender-q 

Собственно, да. Просто другое название блока. И невзначай показанный пример работы с эксчейнджами других типов 🙂

Чистый паблишер и интеграция с FastAPI

from datetime import datetime from uuid import uuid4  from fastapi import FastAPI from mela import Mela from mela.settings import Settings from pydantic import BaseModel  app = FastAPI()  mela_app = Mela(__name__) mela_app.settings = Settings()   class ReportRequest(BaseModel):     start_date: datetime     end_date: datetime     user_id: str     report_id: str | None = None   @app.post("/report") async def read_root(report_request: ReportRequest):     if report_request.report_id is None:         report_request.report_id = str(uuid4())     # some DB writing     publisher = await mela_app.publisher_instance('report-generator')     await publisher.publish(report_request)     return report_request 

application.yml:

connections:   default:     host: localhost     port: 5672     username: user     password: bitnami  publishers:   report-generator:     exchange: report-x     routing_key: new-report 

Тут даже не знаю что комментировать. Разве что добавлю, что в будущем хотелось бы сделать внешний декоратор, чтобы инжектить паблишеры и RPC клиенты в сторонние функции так же, как в следующем примере. Но это уже к планам на будущее. А пока перейдём к следующему примеру.

Инъекции дополнительных паблишеров

Иногда нам нужно реализовать сплиттер или просто по ходу обработки сообщения что-то куда-то запаблишить. Можно, конечно, сделать как в предыдущем разделе, но можно сделать круче:

from datetime import datetime  from pydantic import BaseModel  from mela import Mela from mela.components import Publisher  app = Mela(__name__)   class Document(BaseModel):     text: str     url: str     date: datetime     has_images: bool = False   @app.service('archiver') async def archiver(document: Document, images_downloader: Publisher = 'images-downloader') -> Document:     # archiving document          if document.has_images:         await images_downloader.publish(document)          return document   if __name__ == '__main__':     app.run() 

application.yml:

connections:   default:     host: localhost     port: 5672     username: admin     password: admin  services:   archiver:     consumer:       exchange: archiver-x       routing_key: archiver-q       queue: archiver-q     publisher:       exchange: notify-archived-x       exchange_type: topic       routing_key: document.archived  publishers:   images-downloader:     exchange: images-downloader-x     routing_key: images-downloader-q 

На данный момент инъекции реализованы неправильно с точки зрения типизации. Есть план переписать их на новый служебный тип Annotated. Но… Главное работает.

DLX

Для любого консьюмера поддерживается Dead Letter Exchange.

Сделать его очень просто:

connections:   default:     host: localhost     port: 5672     username: admin     password: admin  services:   service_with_dlx:     consumer:       exchange: dlx-test-x       routing_key: dlx-test-k       queue: dlx-test-q       dead_letter_exchange: dlx-test-dead-letter-x       dead_letter_routing_key: dlx-test-dead-letter-k     publisher:       exchange: test-x       routing_key: test_queue

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

RPC

# server.py import asyncio import aio_pika  from mela import Mela  app = Mela(__name__)   async def fetch(url):     # asynchronously fetching url here and return its body     await asyncio.sleep(1)     return url   @app.rpc_service("fetcher") async def fetcher(url: str):     return {"fetched": await fetch(url)}   bots = {}   def create_bot(bot_id, bot_username, bot_password):     bots[bot_id] = {'username': bot_username, 'password': bot_password}   def get_bot(bot_id):     return bots[bot_id]   @app.rpc_service("bot_manager") async def fetcher(body, message: aio_pika.Message):     if message.headers['method'] == 'create_bot':         create_bot(**body)         return {'result': None, 'status': "OK"}     elif message.headers['method'] == 'get_bot':         return {'result': get_bot(**body), 'status': "OK"}     else:         return {'result': None, 'status': "ERROR_UNKNOWN_METHOD"}  if __name__ == '__main__':     app.run() 

# client.py import asyncio  from mela import Mela  app = Mela(__name__)   async def main():     # RPC calls over RabbitMQ never were simpler!      fetcher = await app.rpc_client_instance("fetcher")      bot_manager = await app.rpc_client_instance("bot_manager")      res = await fetcher.call({'url': "test"})     print(res)      # we can even gather call results!     g = await asyncio.gather(fetcher.call({'url': url1}), fetcher.call({'url': url2}))     print(g)      create_bot_result = await bot_manager.call({         'bot_id': 1,         'bot_username': "LalkaPalka",         'bot_password': "supersecret",     },         headers={'method': 'create_bot'},     )     print(f"create_bot result {create_bot_result}")      get_bot_result = await bot_manager.call({'bot_id': 1}, headers={'method': 'get_bot'})     print(f"get_bot_result {get_bot_result}")      unknown_method_result = await bot_manager.call({'bot_id': 4}, headers={'method': 'getBot'})     print(f"unknown method result: {unknown_method_result}")   if __name__ == '__main__':     url1 = (         'https://tengrinews.kz/kazakhstan_news/vorvalis-dom-izbili-'         'almatinka-rasskazala-zaderjanii-supruga-459127/'     )     url2 = (         'https://www.inform.kz/ru/skol-ko-lichnyh-podsobnyh-'         'hozyaystv-naschityvaetsya-v-kazahstane_a3896073'     )     app.run(main())

connections:   default:     host: localhost     port: 5672     username: user     password: bitnami  rpc-services:   fetcher:     exchange: fetcher-x     routing_key: fetcher-k     queue: fetcher-q     response_exchange: fetching-result-x   bot_manager:     exchange: botmanager-x     routing_key: botmanager-k     queue: botmanager-q     response_exchange: botmanager-result-x

Вот вам пример сразу с двумя не конфликтующими друг с другом RPC-сервисами в одном процессе.

Тут тоже не вижу смысла что-то объяснять, но если будут вопросы — с радостью отвечу.

Несколько подключений и переменные окружения

Очень распространённый кейс — когда нам нужно перекачать данные из одного кластера реббита в другой. Сейчас мы не будем обсуждать правильность этой практики, а просто покажем, как это можно сделать легко и непринуждённо.

# application.yml connections:   input_connection:     host: $RABBIT_INPUT_HOST     port: ${RABBIT_INPUT_PORT|5672}     username: ${RABBIT_INPUT_USERNAME|rabbitmq-bridge}     password: ${RABBIT_INPUT_PASSWORD|rabbitmq-bridge}   output_connection:     host: $RABBIT_OUTPUT_HOST     port: ${RABBIT_OUTPUT_PORT|5672}     username: ${RABBIT_OUTPUT_USERNAME|rabbitmq-bridge}     password: ${RABBIT_OUTPUT_PASSWORD|rabbitmq-bridge}  services:   bridge:     consumer:       connection: input_connection       prefetch_count: ${RABBIT_INPUT_PREFETCH_COUNT|1}       routing_key: ${RABBIT_INPUT_ROUTING_KEY}       exchange: ${RABBIT_INPUT_EXCHANGE}       queue: ${RABBIT_INPUT_QUEUE}     publisher:       connection: output_connection       routing_key: ${RABBIT_OUTPUT_ROUTING_KEY}       exchange: ${RABBIT_OUTPUT_EXCHANGE} 

# app.py from mela import Mela  app = Mela(__name__)   @app.service("bridge") async def serve(body, message):     return body   if __name__ == '__main__':     app.run() 

Как вы можете заметить, код сервиса очень простой. А вот в конфигурационном файле есть кое-что новое. Первое, что бросается в глаза — это переменные окружения. Да, их очень просто сюда вшить. Вот, кстати, пример дотэнва:

RABBIT_INPUT_HOST=localhost RABBIT_INPUT_ROUTING_KEY=routing-key RABBIT_INPUT_EXCHANGE=exchange RABBIT_INPUT_QUEUE=queue RABBIT_OUTPUT_HOST=localhost RABBIT_OUTPUT_PORT=5673 RABBIT_OUTPUT_ROUTING_KEY=routing-key RABBIT_OUTPUT_EXCHANGE=exchange

Всё очень просто и прямолинейно, правда?

А ещё это по сути уже можно упаковывать в докер, и запускать через оркестратор.

Второе, на что нужно обратить внимание в конфиг-файле этого раздела — это то, что здесь два коннекшена, и для консьюмера и паблишера используются разные коннекшены.

На этом у меня юзкейсы закончились. Перейдём к следующему разделу.

Производительность

Цель этой статьи — не похвастаться, да и к тому же производительность фреймворка — это целиком и полностью заслуга авторов aio-pika, а не моя. Детальные бенчмарки — это будет тема для отдельной статьи, но пока просто скажу, что на моём не самом мощном ноуте простой бридж между двумя реббитами обрабатывает порядка 500 сообщений в секунду. А лучшее, чего мне удавалось добиться от обычной pika — это 80-100 сообщений в секунду. В случае с фетчингом страниц по всем понятным причинам не асинхронная pika не могла показать вообще сколько-нибудь адекватный результат.

Заключение

Как мне кажется, мне удалось составить неплохую внешнюю апишку. За исключением инъекций, но про них я писал выше. Внутри — бардак. Всё работает, конечно, но мейнтейнить и дальше развивать фреймворк будет сложно. Я уже начал переписывать ядро в чистовой вариант, но мне не хватает мотивации.

Поэтому я и написал эту статью. Если статья получит отклик, если этот фреймворк кого-то заинтересует, если найдутся люди, которые будут им пользоваться, а может даже и контрибьютить, то я обязательно продолжу работу над ним. Можете даже просто звёздочек накидать, это тоже даст мне хоть какую-то обратную связь 🙂

В конце дублирую ссылку на GitHub

Пишите вопросы в комментах. Ответами на самые важные буду дополнять статью, на остальные отвечу просто in-place.