youeduc-sdk-messaging

SDK Python de mensageria assíncrona orientada a eventos sobre Apache Kafka, construída sobre aiokafka e asyncio. Publica eventos validados por JSON Schema e os consome com retry por tópicos, DLQ, deduplicação, commit manual de offsets e observabilidade (OpenTelemetry + logs estruturados com structlog).

• Distribuição: youeduc-sdk-messaging; import: sdk_messaging.
• Tipada (py.typed), Python >= 3.10, Kafka >= 3.x.
• Existe uma SDK Node equivalente que segue o mesmo protocolo (tópicos, headers, retry/DLQ e telemetria): um serviço Python publica e um serviço Node consome, e o contrário.

Instalação

Fixe sempre uma versão exata:

pip install "youeduc-sdk-messaging==X.Y.Z"

Com a deduplicação compartilhada em Redis (sdk_messaging.integrations.redis):

pip install "youeduc-sdk-messaging[redis]==X.Y.Z"

Conferir a versão instalada:

python -c "import sdk_messaging; print(sdk_messaging.__version__)"

Dependências de runtime: aiokafka, jsonschema, referencing, structlog e opentelemetry-api. O opentelemetry-sdk e os exporters ficam a cargo da aplicação. Django, Flask, FastAPI e Celery não são dependências: os adaptadores sdk_messaging.integrations.django e sdk_messaging.integrations.flask só importam o framework quando o próprio serviço os importa.

Publicar

import asyncio
from pathlib import Path

from sdk_messaging import FileSystemSchemaLoader, KafkaConfig, Messaging


async def main() -> None:
    async with Messaging(
        KafkaConfig.from_env(),                      # KAFKA_BOOTSTRAP_SERVERS etc.
        service="enrollment-api",                    # kebab-case; vai no header `producer`
        environment="production",
        schema_loader=FileSystemSchemaLoader(Path("schemas")),
    ) as messaging:
        result = await messaging.publish(
            "enrollment", "student", "enrolled", "v1",   # domain, aggregate, event, version
            {"student_id": "s-123", "course_id": "c-9"},
            key="s-123",                                 # define a partição (ordem por chave)
            idempotency_key="enrollment-987",            # opcional; chave de dedup preferida
        )
        print(result.message_id, result.topic, result.partition, result.offset)


asyncio.run(main())

O evento vai para o tópico enrollment.events.v1 com event_type = "student.enrolled" e é validado contra schemas/enrollment/student-enrolled/v1.json (JSON Schema Draft 2020-12) antes de sair. publish aguarda o ack do broker; payload inválido levanta SchemaValidationError sem enviar nada, e erro do Kafka levanta PublishError.

Consumir

import asyncio
import signal
from pathlib import Path

from sdk_messaging import (
    ConsumedMessage,
    FileSystemSchemaLoader,
    KafkaConfig,
    Messaging,
    PermanentProcessingError,
)


async def on_enrolled(message: ConsumedMessage) -> None:
    if "course_id" not in message.payload:
        raise PermanentProcessingError("course_id ausente")  # vai direto para a DLQ
    ...  # retornar = sucesso; qualquer outra exceção = retry


async def main() -> None:
    stop = asyncio.Event()
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(sig, stop.set)

    async with Messaging(
        KafkaConfig.from_env(),
        service="learning-worker",  # também é o consumer group padrão
        environment="production",
        schema_loader=FileSystemSchemaLoader(Path("schemas")),
    ) as messaging:
        consumer = messaging.consumer(
            domain="enrollment",
            version="v1",
            handlers={"student.enrolled": on_enrolled},  # event_type -> handler async
            processing_timeout_s=60,
        )
        await consumer.start()
        consumer_done = asyncio.ensure_future(consumer.wait())
        signal_received = asyncio.ensure_future(stop.wait())
        await asyncio.wait({consumer_done, signal_received}, return_when=asyncio.FIRST_COMPLETED)
        signal_received.cancel()
    # Ao sair do bloco, os consumers terminam o registro em voo, commitam e o producer fecha.
    await consumer_done  # relança o erro fatal que derrubou o consumer, se houve


asyncio.run(main())

Eventos do domínio sem handler são ignorados e o offset avança. Os tópicos (principal, retry e DLQ do grupo) precisam existir antes do consumer iniciar: crie-os por IaC ou, em desenvolvimento, com messaging.ensure_topics(...).

Garantias

Não há exactly-once nem transações Kafka.

Retry e DLQ

A política padrão (RetryPolicy()) reprocessa uma falha transitória após 30 s, 5 min, 1 h e 12 h; a quinta falha vai para a DLQ. Cada atraso tem um tópico de retry por consumer group ({tópico}.{grupo}.retry.30s ... .retry.12h) e a DLQ é {tópico}.{grupo}.dlq. A mensagem de retry é relida só pelo próprio grupo e nunca volta ao tópico principal; enquanto espera, a partição de retry fica pausada sem bloquear as demais.

• PermanentProcessingError (ou subclasse), payload inválido e falha de schema vão direto para a DLQ.
• Qualquer outra exceção, inclusive timeout de processamento, segue a política de retry.
• Política própria: RetryPolicy(delays_s=(10, 60, 300)); política vazia desativa o retry.
• A DLQ nunca marca a deduplicação, então replay_dlq(...) pode reprocessar as mensagens.

Prefixo de ambiente

Para homologação e produção no mesmo cluster, resource_prefix (kebab-case, como hmg ou prd) vira o primeiro segmento do tópico principal e do consumer group, e por consequência de retry, DLQ e replay:

Messaging(kafka, service="catalogo-api", environment="homologacao", schema_loader=loader, resource_prefix="hmg")
# tópico: hmg.usuarios.events.v1   grupo: hmg.catalogo-api
# retry:  hmg.usuarios.events.v1.catalogo-api.retry.30s   DLQ: hmg.usuarios.events.v1.catalogo-api.dlq

Prefixar também o grupo impede que réplicas dos dois ambientes dividam as partições do mesmo group.id. Sem prefixo, os nomes não mudam.

Deduplicação

InMemoryDuplicateChecker serve a um processo só. Para várias réplicas, use o store Redis (extra [redis]), com um namespace por sistema e ambiente:

from sdk_messaging.integrations.redis import RedisDedupStore

store = RedisDedupStore.from_url("redis://redis:6379/0", namespace="hmgcatalogo_api")
consumer = messaging.consumer(
    domain="enrollment",
    version="v1",
    handlers={"student.enrolled": on_enrolled},
    duplicate_checker=store.checker("learning-worker"),
)

A chave é marcada só depois do sucesso do handler; retry e DLQ não marcam.

Contratos embutidos

Os schemas e tipos (TypedDict) dos eventos corporativos vêm no pacote, na mesma versão da SDK:

from sdk_messaging.contracts import schema_loader
from sdk_messaging.contracts.usuarios import UsuarioDesativado

loader = schema_loader()  # InMemorySchemaLoader com todos os schemas embutidos

Para combinar com schemas próprios do serviço: ChainSchemaLoader(FileSystemSchemaLoader(Path("schemas")), schema_loader()).

Configuração

KafkaConfig.from_env() lê KAFKA_BOOTSTRAP_SERVERS (obrigatória), KAFKA_CLIENT_ID, KAFKA_SECURITY_PROTOCOL (PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL), KAFKA_SASL_MECHANISM, KAFKA_SASL_USERNAME e KAFKA_SASL_PASSWORD. Django, Flask e serviços sem framework também podem montar tudo a partir de um mapa SDK_MESSAGING (SERVICE, ENVIRONMENT, KAFKA, REDIS, RESOURCE_PREFIX...) com sdk_messaging.integrations.runtime.MessagingSettings.from_mapping.

Licença

MIT.