Набор компонентов реализующих очередь необработанных сообщений (Dead Letter Queue).

Пример подключения

testapp/dlq/apps.py:

from django.apps import AppConfig as AppConfigBase


class AppConfig(AppConfigBase):
    name = __package__

    def ready(self):
        self._configure_dlq()

    def _configure_dlq(self):
        from testapp.core import bus
        from explicit.dlq.config import DLQConfig, configure_dlq
        # реализация репозитория специфичная для используемого слоя хранения данных
        from testapp.dlq.adapters.db import Repository
        
        # реализация обработчика сообщений, специфичного для приложения
        from testapp.dlq.services.dispatch import dispatch_message

        config = DLQConfig(
            bus=bus,
            repository=Repository(),
            message_dispatcher=dispatch_message
        )
        # регистрация репозитория и регистрация обработчиков команд
        configure_dlq(config)

testapp/dlq/services/dispatch.py:

import json
from explicit.contrib.messagebus.event_registry import Message

from testapp.core import bus
from testapp.core import event_registry

def dispatch_message(raw_message_value: bytes, topic: str):
    """Преобразует сообщение в событие и отправляет его в шину."""
    message = json.loads(raw_message_value)

    if event := event_registry.resolve(
        Message(
            topic=topic,
            type=message.get('type'),
            body=message
        )
    ):
        bus.handle(event)

testapp/entrypoints/kafka.py:

def bootstrap() -> None:
    from explicit.dlq.infrastructure.handlers import RegisterInDLQOnFailure

    from testapp.core import bus

    from testapp.dlq.services.dispatch import dispatch_message
    topics = ('test.foo.topic',)

    for raw_message in adapter.subscribe(*topics):
        with RegisterInDLQOnFailure(
            bus=bus,
            topic=raw_message.topic(),
            raw_message_value=raw_message.value(),
            raw_message_key=raw_message.key
        ):
            dispatch_message(raw_message.value(), raw_message.topic())

Готовые компоненты (contrib)

В пакете реализованы готовые к использованию компоненты:

• Реализация абстрактного хранилища сообщений на базе Django ORM
• REST API для работы с хранилищем сообщений. Предназначен для совместной работы с реализацией хранилища сообщений django ORM.

Запуск тестов

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

Для корректного запуска тестов необходимы следующие значения переменных окружения в системе. Вместо <login> и <password> необходимо подставить данные от УЗ БАРС:

UV_INDEX="internal=http://<login>:<password>@nexus.py.bars.group/repository/pypi-all/simple"
UV_DEFAULT_INDEX="https://pypi.org/simple"
UV_INDEX_STRATEGY=unsafe-best-match
# Для docker 29+
TOX_DOCKER_GATEWAY=localhost

Чтобы не менять значения системных переменных, можно создать .env файл c переменными окружения и запускать команды через python-dotenv.

Установка tox с необходимыми зависимостями:

uv tool install tox --with tox-uv --with m3-tox-docker

Или с использованием python-dotenv:

dotenv run uv tool install tox --with tox-uv --with m3-tox-docker

Запуск тестов:

tox

Через python-dotenv:

dotenv run tox