Metadata-Version: 2.5
Name: youeduc-sdk-messaging
Version: 0.6.0
Summary: SDK de mensageria assíncrona orientada a eventos sobre Apache Kafka (aiokafka)
Author: YouEduc
License-Expression: MIT
License-File: LICENSE
Keywords: aiokafka,asyncio,dlq,events,json-schema,kafka,messaging,retry
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Distributed Computing
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: aiokafka<0.15,>=0.14.0
Requires-Dist: jsonschema>=4.23.0
Requires-Dist: opentelemetry-api>=1.27.0
Requires-Dist: referencing>=0.28.4
Requires-Dist: structlog>=24.4.0
Provides-Extra: dev
Requires-Dist: mypy==2.3.1; extra == 'dev'
Requires-Dist: opentelemetry-sdk==1.44.0; extra == 'dev'
Requires-Dist: pytest-asyncio==1.4.0; extra == 'dev'
Requires-Dist: pytest==9.1.1; extra == 'dev'
Requires-Dist: redis==8.1.0; extra == 'dev'
Requires-Dist: ruff==0.16.8; extra == 'dev'
Requires-Dist: types-jsonschema==4.26.0.20260518; extra == 'dev'
Provides-Extra: redis
Requires-Dist: redis>=5.0.1; extra == 'redis'
Description-Content-Type: text/markdown

# 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:

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

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

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

Conferir a versão instalada:

```bash
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

```python
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

```python
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

| Garantia | Como |
|---|---|
| **At-least-once** no consumo | O offset só é commitado depois do desfecho final (sucesso, DLQ, ignorada) ou do envio **confirmado** para retry/DLQ. Um handler pode rodar mais de uma vez para a mesma mensagem: **handlers devem ser idempotentes** ou usar um `DuplicateChecker`. |
| **Ordem por chave** | Mensagens com a mesma `key` vão para a mesma partição e são processadas em sequência. Uma mensagem que vai para retry sai da ordem. |
| **Producer idempotente** | Um producer por processo, `acks=all` e idempotência ligada, compartilhado por publish, retry e DLQ. |
| **Schema obrigatório** | Todo payload é validado no publish e no consumo. |
| **Sem PII em logs** | Logs, spans, métricas e erros de schema nunca incluem valores do payload nem a `key`. |

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:

```python
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:

```python
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:

```python
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.
