Guia do kit do adaptador

Este guia leva você de um repositório vazio até um adaptador que declara suas fontes, toma trabalho da fila, traduz o que a fonte devolve e entrega ao Backend, provado pela suíte de conformidade do kit. Você não precisa ler o código do kit: cada passo mostra o trecho do adaptador de exemplo que o faz, com o nome do arquivo de onde ele veio, e um teste do kit confere que o trecho continua lá.

O exemplo mora em examples/example_adapter/, ao lado deste guia, e é escrito só sobre o kit para uma fonte fictícia: o Livro de Registro e o Arquivo Central. Ele roda no portão do kit, então uma mudança no kit que o quebre derruba o portão. Quando o guia disser "o exemplo", é esta pasta. Os caminhos dos trechos partem da pasta do kit, kits/adapter/; os de docs/ e adapters/ partem da raiz do repositório.

Para a visão de processo (como uma instituição entra no produto, do primeiro contato à instalação), leia antes apps/docs/content/integrar/playbook-novo-cliente.md. Este guia é o passo a passo do código. Se a sua fonte não tem documentação nem padrão, comece pelo começo rápido e siga para "Fonte sem padrão".

O que o kit é e o que não é

Ao fim desta seção você sabe o que escrever e o que já vem pronto.

Um adaptador é o programa da instituição que busca os feitos na fonte dela e os entrega ao Backend do produto, no formato que o contrato publica. O kit é o cliente de referência que o produto mantém para isso (AD-020). Ele faz o trabalho repetido, e deixa para você só o que é da fonte.

Quem	O que faz
O Backend (produto)	Guarda o acervo, mantém a fila de ingestão, confere cada mensagem contra o contrato e assina o ticket de envio de uma peça
O kit	Cliente do Backend, laço que toma trabalho e roteia pela chave da fonte, envio da peça por ticket, ritmo adaptativo, pausa por Retry-After, configuração comum e suíte de conformidade
Você, a instituição	O cliente de cada fonte, o mapeador de cada fonte, o vocabulário (a tradução das palavras da fonte), a declaração e a imagem

O kit também não é algumas coisas:

• Não fala com o banco do Backend. Toda escrita passa pela porta de ingestão, e o arquivo de uma peça só vai ao armazenamento de objetos pelo ticket que o Backend assina.
• Não conhece fonte nem instituição. Quem diz o que cada palavra da fonte significa é o seu vocabulário.
• Não é obrigatório. O contrato é o esquema JSON que o Backend publica (AD-007). Um time em outra linguagem implementa pelo esquema e usa os cenários da suíte de conformidade como prova.

A vida de um trabalho da fila

O desenho abaixo mostra quem faz o quê. As caixas marcadas com "(seu)" são as que você escreve. Repare que o cliente e o mapeador nunca se falam: o laço do kit leva a resposta de um ao outro.

Em ordem de tempo, o que acontece:

Dois detalhes importam. O kit só toma trabalho depois que o Backend aceita a declaração, e insiste com esperas crescentes enquanto ele não aceita. E o laço é um conjunto contínuo de vagas, não um lote por vez: a vaga que libera pega outro trabalho na hora, e um trabalho lento ocupa só a vaga dele.

Começo rápido

Ao fim desta seção você tem um adaptador declarado, que toma trabalhos e entrega envelopes que o contrato aceita, tudo contra o Backend falso do kit, sem rede e sem Docker.

Antes de começar, confira o que precisa estar instalado:

Item	Versão	Observação
Python	3.14.7	Exata: o kit e o exemplo fixam a versão
uv	0.12.18	Exata, também em required-version
Acesso ao feed proceedings	leitura	Só para instalar o kit publicado; o README.md do kit explica como

1. Veja o exemplo funcionar. Da pasta do kit:

cd kits/adapter
uv sync --locked
uv run pytest -q examples/example_adapter/tests

Todos os testes passam, sem rede. O mais importante é o de conformidade, test_example_conformance.py: ele sobe o adaptador inteiro contra o Backend falso do kit e uma fonte simulada.

2. Crie o projeto do seu adaptador. Copie a pasta examples/example_adapter/ para adapters/<instituição>/ e troque os nomes: a pasta src/example_adapter/, o name do projeto e os import que a citam. O adaptador é um projeto independente: tem o pyproject.toml, o uv.lock e o ambiente dele, e consome o kit publicado, nunca por caminho. O pyproject.toml do exemplo já é o de um projeto completo, e vem junto com a cópia. Ele está inteiro abaixo, porque cada parte faz falta:

examples/example_adapter/pyproject.toml

# Projeto próprio do adaptador: tem o lock, o ambiente e o portão dele, e consome o kit publicado,
# nunca por caminho. Toda versão é exata. Dentro do kit, o portão só o confere contra o guia; ao
# copiar a pasta para `adapters/<instituição>/`, ele é o `pyproject.toml` do projeto novo.

[project]
name = "example-adapter"
version = "0.1.0"
description = "Adaptador de exemplo: uma fonte fictícia, só sobre o kit."
requires-python = "==3.14.7"
dependencies = [
    "httpx==0.28.1",
    "proceedings-adapter-kit==0.3.4",
    "pydantic==2.13.5",
    "pydantic-settings==2.15.0",
]

[tool.uv]
required-version = "==0.12.18"
link-mode = "copy"

# O kit vem do feed da Xertica, e o índice só serve o kit: o resto vem do PyPI. A credencial vem do
# ambiente, `UV_INDEX_XERTICA_USERNAME` e `UV_INDEX_XERTICA_PASSWORD`, nunca de arquivo.
[[tool.uv.index]]
name = "xertica"
url = "https://pkgs.dev.azure.com/XerticaBR/mprs-fdm/_packaging/proceedings/pypi/simple/"
explicit = true

[tool.uv.sources]
proceedings-adapter-kit = { index = "xertica" }

# O extra `testing` do kit traz a suíte de conformidade; `pytest-asyncio` roda os testes `async`.
[dependency-groups]
dev = [
    "mypy==2.3.1",
    "proceedings-adapter-kit[testing]==0.3.4",
    "pytest==9.1.1",
    "pytest-asyncio==1.4.0",
    "ruff==0.16.8",
]

# Sem isto o pacote não se instala, e os testes e a imagem não o acham.
[build-system]
requires = ["uv_build==0.12.18"]
build-backend = "uv_build"

[tool.ruff]
target-version = "py314"
line-length = 100

[tool.ruff.lint]
# A lista de um adaptador de verdade, sem a família de simplificação: o código dela é uma palavra
# que o guarda de nomes do kit proíbe no exemplo.
select = ["E", "W", "F", "I", "N", "UP", "B", "A", "RUF", "ASYNC", "S", "PT", "TCH"]

[tool.ruff.lint.flake8-type-checking]
# Pydantic lê as anotações em tempo de execução
runtime-evaluated-base-classes = ["pydantic.BaseModel", "pydantic_settings.BaseSettings"]
exempt-modules = ["typing"]

[tool.ruff.lint.per-file-ignores]
# assert é a forma do pytest
"**/tests/**" = ["S101", "S105", "S106"]  # assert do pytest; credenciais descartáveis de teste

[tool.mypy]
python_version = "3.14"
strict = true
warn_return_any = true
files = ["."]
exclude = ["^\\.venv/"]

[tool.pytest.ini_options]
asyncio_mode = "auto"
addopts = "--import-mode=importlib"
testpaths = ["tests"]
# Os testes rodam também sem instalar o pacote, direto desta pasta.
pythonpath = ["src"]

O que cada parte faz, para você não tirar nada por engano:

Parte	Para que serve
dependencies e [tool.uv.sources]	O kit entra pinado, e o índice da Xertica, explícito, só serve o kit. Toda versão é exata, sem ^, ~ nem latest
[build-system]	Sem ele o pacote não se instala, e os testes e a imagem não o acham
[dependency-groups]	O extra testing do kit traz a suíte de conformidade, e o pytest-asyncio roda os testes async
asyncio_mode = "auto"	Sem ele, os testes de conformidade, que são async, não rodam como teste
[tool.ruff] e [tool.mypy]	O portão do adaptador: estilo e tipagem estrita
pythonpath = ["src"]	Deixa o pytest achar o pacote antes de instalá-lo

A pasta do exemplo traz também um ruff.toml. Ele existe só para o exemplo rodar dentro do kit: estende a configuração do kit por um caminho que não existe no seu projeto. Apague-o, e vale o [tool.ruff] do pyproject.toml acima.

O kit vem do feed da Xertica, então exporte as credenciais antes de instalar. O token vale cerca de uma hora e nunca vai para arquivo nem para o repositório:

export UV_INDEX_XERTICA_USERNAME=aad
export UV_INDEX_XERTICA_PASSWORD="$(az account get-access-token --resource 499b84ac-1321-427f-aa17-267ca6975798 --query accessToken -o tsv)"

Quem não tem login no tenant da Xertica usa, no lugar do comando do az, um token de acesso pessoal (PAT, do inglês personal access token) com leitura de pacotes, que o time do feed concede. O nome de usuário não importa.

3. Instale e rode os testes do seu projeto. Da pasta do seu adaptador:

rm ruff.toml
uv sync
uv run pytest -q

O primeiro uv sync cria o uv.lock, e você o versiona. Se os testes passam aqui, a cópia está de pé, e o resto do guia é trocar o que é do exemplo pelo que é da sua fonte.

4. Escreva as cinco peças, nesta ordem. Cada uma tem a sua seção mais adiante; aqui só o que cada uma é:

Peça	Arquivo do exemplo	O que é
Configuração	settings.py	As variáveis do kit mais o endereço e a credencial da sua fonte
Declaração	declaration.py	As fontes que você atende, a versão do adaptador e os tipos de peça próprios
Cliente	sources/<sistema>/client.py	Só E/S: busca o feito e o arquivo da peça
Mapeador	sources/<sistema>/mapper.py	Função pura: a resposta da fonte vira o envelope
Montagem	__main__.py	Registra uma fonte por chave declarada e chama o run do kit

A montagem é a mais curta, e mostra como as peças se encaixam:

examples/example_adapter/src/example_adapter/__main__.py

def build_adapter(
    settings: ExampleSettings, http: httpx.AsyncClient, source_http: httpx.AsyncClient
) -> Adapter:
    """Uma fonte por chave declarada. O login próprio do Arquivo usa o cliente `http` comum."""
    vocabulary = Vocabulary.load()
    return Adapter(
        declaration(),
        {
            "ledger": Source(
                LedgerClient(settings, source_http),
                partial(map_ledger, vocabulary=vocabulary, version=ADAPTER_VERSION),
            ),
            "archive": Source(
                ArchiveClient(settings, source_http, ArchiveLogin(settings, http)),
                partial(map_archive, vocabulary=vocabulary, version=ADAPTER_VERSION),
            ),
        },
    )


async def serve(settings: ExampleSettings) -> None:
    async with httpx.AsyncClient() as http, source_http_client(settings) as source_http:
        await run(build_adapter(settings, http, source_http), settings)

5. Prove com a suíte de conformidade. O teste roda o seu adaptador, o de verdade, pelo laço do kit contra o Backend falso, que valida cada mensagem pelos esquemas do contrato:

examples/example_adapter/tests/test_example_conformance.py

async def test_the_example_conforms_to_the_published_contract_from_declaration_to_each_failure(
    settings: ExampleSettings, payload: Callable[[str], bytes], run_adapter: RunAdapter
) -> None:
    adapter = _adapter_over(SimulatedSources(payload), settings)

    result = await run_adapter(
        adapter,
        settings=settings,
        jobs=[
            _job(1, "LDG-2024-0001"),
            _job(2, "ARQ-0077", "archive"),
            _job(
                3,
                "LDG-2024-0001",
                kind="filing",
                filing_id=FILING_ID,
                filing_source_identifier="f-1",
            ),
            _job(4, "LDG-404"),
            _job(5, "LDG-503"),
            _job(6, "LDG-400"),
            _job(7, "LDG-BAD"),
            _job(8, "ARQ-GONE", "archive"),
        ],
    )

    result.assert_conforming()

result.assert_conforming() falha listando a regra e o motivo se qualquer mensagem foi recusada. Passou? O seu adaptador fala o contrato.

6. Leia a recusa quando ela vier. Cada recusa nomeia a regra que a causou. Se o vocabulário mandasse um conceito que o Backend não publica, o teste falharia assim:

AssertionError: o adaptador violou o contrato: concept-key: chave de conceito fora das publicadas: role:polo-passivo

A seção sobre as regras lista todas e diz como corrigir cada uma.

As pastas e o papel de cada arquivo

Ao fim desta seção você sabe onde cada coisa mora, e o que nunca deve morar ali.

adapters/<instituição>/
  Dockerfile                    imagem do adaptador
  pyproject.toml, uv.lock       projeto próprio, versões exatas
  installation/compose.yaml     o que a instituição acrescenta ao compose do produto
  src/<nome>_adapter/
    __main__.py                 monta o Adapter e chama o run do kit
    declaration.py              fontes, versão e tipos de peça próprios
    settings.py                 as variáveis do kit mais as da fonte
    version.py                  a versão do adaptador
    envelope.py                 peças puras que os mapeadores dividem
    vocabulary/                 o vocabulário, em dado (terms.json)
    sources/response.py         o que o cliente devolve e o mapeador lê
    sources/<sistema>/client.py E/S, pelo SourceHttp do kit
    sources/<sistema>/mapper.py função pura, resposta da fonte para envelope
    sources/<sistema>/login.py  só se a fonte tem login próprio
  tests/                        mapeadores, clientes e conformidade

Arquivo	Faz	Não faz
client.py	Monta a rota, põe a credencial, chama SourceHttp.get e devolve a resposta	Traduz, conhece o vocabulário, monta envelope
mapper.py	Lê a resposta, consulta o vocabulário, devolve o envelope	Rede, arquivo, relógio, qualquer E/S
vocabulary/	Guarda as palavras da fonte e o conceito de cada uma, em dado	Mora no Backend: o Backend nunca conhece a tradução de uma fonte
declaration.py	Diz ao Backend as fontes, a versão e os tipos próprios	Envia a declaração: isso é do kit
__main__.py	Liga cliente, mapeador e vocabulário, e entrega ao run	Regra de negócio

A separação entre cliente e mapeador é a mais importante do desenho. O mapeador não faz E/S, então se testa com respostas gravadas, sem rede. O cliente não traduz, então se testa com uma fonte simulada. O exemplo tem um teste que recusa o contrário: um cliente que importa o vocabulário, ou um mapeador que importa a rede, falha o portão, nomeando o arquivo.

A configuração

Ao fim desta seção você sabe quais variáveis o kit lê, e como acrescentar as da sua fonte.

O kit lê só o ambiente do processo, e valida tudo na subida: variável ausente ou fora do limite impede o adaptador de subir, e a mensagem nomeia o campo. Segredo é SecretStr, que não aparece nos registros nem em repr.

Variável	Obrigatória	Padrão	Para que serve
BACKEND_URL	sim		Endereço do Backend
INGESTION_SERVICE_TOKEN	sim		O mesmo segredo que o Backend lê em INGESTION_SERVICE_TOKEN
RATE_LIMIT_PER_SECOND	sim		Teto do ritmo de pedidos à fonte. Não tem padrão: a fonte não publica limite, e a instalação decide depois de medir
SOURCE_TIMEZONE	sim		Nome do fuso (por exemplo America/Sao_Paulo) em que a fonte grava datas sem fuso
CONCURRENCY	não	4	Vagas em voo, de 1 a 64
BATCH_SIZE	não	10	Máximo de trabalhos por pedido de lote à fila, de 1 a 100
IDLE_SECONDS	não	2.0	Espera quando não há vaga nem trabalho
REQUEST_TIMEOUT_SECONDS	não	300.0	Tempo limite de cada leitura de um pedido, à fonte e ao Backend (a conexão tem 5 s, à parte). Uma fonte pode levar minutos para começar a mandar um documento grande
WORKER	não	servidor/pid	Nome com que o laço se identifica ao tomar trabalho

Para acrescentar as variáveis da sua fonte, herde de KitSettings. Cada campo vira uma variável de ambiente com o mesmo nome, em maiúsculas:

examples/example_adapter/src/example_adapter/settings.py

class ExampleSettings(KitSettings):
    """Segredo é `SecretStr`: não vaza em log nem repr."""

    ledger_base_url: AnyHttpUrl
    ledger_api_key: SecretStr

    archive_base_url: AnyHttpUrl
    archive_token_url: AnyHttpUrl
    archive_client_id: str = Field(min_length=1)
    archive_client_secret: SecretStr

O exemplo tem um teste que lê o ambiente, confere os valores, e recusa a subida sem uma credencial, nomeando-a (test_example_shape.py). Faça o mesmo com as suas.

O envelope por dentro

Ao fim desta seção você sabe, objeto por objeto, o que o envelope exige e o que aceita, e o que o Backend faz quando o sigilo não vem.

O envelope é a mensagem que o seu adaptador entrega para cada trabalho da fila. Ele leva duas coisas: a resposta da fonte, do jeito que chegou (a captura), e o processo já traduzido para o vocabulário do produto. É o que o mapeador devolve, e o que o Backend confere campo a campo contra o contrato.

O desenho mostra as peças e onde o termo, a palavra da fonte junto com o conceito, aparece:

A tabela lista todos os objetos do contrato, com o nome que ele usa. Obrigatório é o mínimo para o Backend aceitar o objeto. Opcional é o que a fonte pode dizer, e que fica de fora quando ela não diz. O Backend recusa o campo que o contrato não conhece.

Objeto	Obrigatório	Opcional
O envelope (Envelope)	contract_version, adapter_version, source_system, capture, proceeding	job_id, notice_id
A captura, o bruto da fonte (CaptureIn)	source_number, endpoint, content_type, content, digest	parameters
O processo (ProceedingIn)	source_number	claim_value, confidentiality, confidentiality_rank, court, docketed_at, electronic, events, filed_at, filings, flags, identifiers, investigation_type, jurisdiction_level, jurisdiction_scope, links, locality_ref, main_subject, panel, parties, priorities, procedural_class, procedural_track, size_bytes, status, subjects, title
O termo, a palavra da fonte e o conceito (SourceTermRef)	vocabulary	code, concept, label
A parte (PartyIn)	ref, role	addresses, anonymous, document_number, identifications, legal_aid, name, person_kind, receives_notice, represents_ref, sex
O evento (EventIn)	ref, occurred_at, source_label	deadline, details, moved_by, movements, references, responsible, reversed, sequence
A peça (FilingIn)	source_identifier, type	confidentiality, confidentiality_rank, created_at, declared_mimetype, event_refs, label, signed, size_bytes, source_digest
O órgão julgador (CourtIn)	code, name	instance, locality_ref, registry_code
O endereço da parte (AddressIn)	nenhum	city, complement, country, district, number, postal_code, state, street
O documento de identificação da parte (IdentificationIn)	type, number	issuer
O prazo de um evento (DeadlineIn)	nenhum	days, party, status
O instituto de sim ou não (FlagIn)	concept, value	nenhum
O identificador do processo (IdentifierIn)	scheme, value	nenhum
O vínculo com outro processo (LinkIn)	linked_number, type	linked_source_system
O assunto (SubjectIn)	term	main

Cinco coisas que a tabela não diz:

• O processo mínimo tem só o source_number. É aceito, e o exemplo o faz quando a fonte devolve só o número. As listas de partes, eventos e peças, quando faltam, valem como vazias.
• O termo (SourceTermRef) exige no contrato só o vocabulário, mas o Backend exige também a palavra original, em code ou em label. O que é opcional de verdade é o concept: sem ele, o valor fica "sem tradução".
• contract_version é a versão do contrato que o envelope fala. Hoje é 2, e o exemplo a tira da constante CONTRACT_VERSION, que o kit publica.
• job_id é opcional no contrato, mas o adaptador que atende um trabalho da fila o manda: o Backend fecha o trabalho na mesma transação que grava o processo.
• notice_id liga um aviso ao processo. O exemplo não o usa.

Atenção: sigilo ausente é sigilo máximo. Processo sem confidentiality_rank entra com o nível 32767, o máximo da escala, e a peça sem o nível dela também: ela não herda o do processo. O sigilo falha fechado, e quem tem o teto abaixo do máximo não enxerga o que chegou sem nível. Por isso, nunca mande "público" porque o dado faltou. Mande o nível que a fonte disse, ou nada. O exemplo manda o nível do processo e não o das peças, então as peças dele entram no máximo; se a sua fonte diz o sigilo de cada peça, mande o confidentiality_rank da peça.

O que o Backend confere além do esquema

O esquema JSON do contrato diz a forma. O Backend confere mais seis coisas antes de gravar, e a suíte de conformidade as confere com o mesmo nome: um envelope que passa na suíte passa também no Backend, e o que a suíte recusa o Backend recusa, com 422. Um corpus de casos, que a suíte julga no portão do kit e o modelo do Backend julga no portão da plataforma, mantém as duas iguais. Cubra estas também no teste do mapeador, porque a suíte só vê as respostas que você gravou.

• term-code-or-label: todo termo traz code ou label, e a string vazia conta como ausente.
• party-name-or-document: toda parte traz name ou document_number, e a string vazia conta como ausente.
• timezone: toda data traz o fuso (filed_at, docketed_at, occurred_at e created_at), e data sem fuso é recusada. O Backend lê a data como ISO 8601 estendido, com o deslocamento completo (-03:00) ou Z: +03, 20240304T101500Z, 2024-W10-1T10:15:00Z e 24:00 são recusados. O mapeador aplica o SOURCE_TIMEZONE e escreve a data nesse formato.
• reference: as referências fecham. Cada event_refs de uma peça cita um evento do mesmo processo, e cada represents_ref cita uma parte dele. O exemplo descarta a referência a um evento que a fonte não mandou.
• capture-base64: o content da captura é base64.
• capture-digest: o digest (sha256 ou sha512) confere com o conteúdo decodificado.

Qual vocabulário cada campo de termo leva

O contrato aceita qualquer vocabulário em qualquer termo, mas a lista a que o conceito pertence é a do campo. Termo no vocabulário errado cita uma chave que a lista não tem, ou grava a palavra na lista errada. A tabela diz, para cada campo que leva um termo, o vocabulário certo. Um caminho como proceeding.parties[].role é o role de cada item de parties.

Campo do termo	Vocabulário
proceeding.procedural_class	procedural_class
proceeding.main_subject	subject
proceeding.subjects[].term	subject
proceeding.priorities[]	priority
proceeding.procedural_track	procedural_track
proceeding.jurisdiction_level	jurisdiction_level
proceeding.jurisdiction_scope	jurisdiction_scope
proceeding.investigation_type	investigation_type
proceeding.status	status
proceeding.confidentiality	confidentiality
proceeding.court.instance	instance
proceeding.panel.instance	instance
proceeding.links[].type	link_type
proceeding.parties[].role	role
proceeding.parties[].sex	sex
proceeding.parties[].person_kind	person_kind
proceeding.parties[].identifications[].type	document_type
proceeding.events[].movements[]	movement
proceeding.events[].deadline.status	deadline_status
proceeding.filings[].type	filing_type
proceeding.filings[].confidentiality	confidentiality

Quando mandar code e quando label

Um termo precisa de um dos dois, ou dos dois, e a string vazia não conta. Mande o code quando a fonte tem um código estável para a palavra. Mande o label quando a fonte só escreve o texto. Mande os dois quando ela dá os dois. Não mande espaços no lugar de um valor que faltou: o Backend os apara, e o termo ficaria sem palavra nenhuma.

O Backend registra cada combinação de code e label como um termo só. A mesma palavra da fonte deve ir sempre com os mesmos campos: code: "DIST" numa resposta e code: "DIST" com label: "Distribuído" em outra viram dois registros. O exemplo faz isto no vocabulário: a consulta trata a string vazia como ausente e devolve a palavra original que recebeu.

O formato do valor da causa

O claim_value é texto decimal, com ponto antes das casas e sem separador de milhar. O contrato recusa o texto que não segue esse formato, então a fonte brasileira, que escreve vírgula nas casas e ponto no milhar, precisa de conversão no mapeador. Valor ausente, ou zero, fica de fora, porque a fonte não o afirmou.

A fonte escreve	O envelope leva
1.500,50	1500.50
12.000,00	12000.00

Do payload da fonte ao envelope

Ao fim desta seção você acompanha um processo do Livro de Registro, do JSON que a fonte devolve até o envelope que o Backend recebe, e sabe de onde vem cada campo.

O caminho tem cinco passos: a resposta gravada, a tabela que liga cada campo dela ao envelope, a tabela que traduz cada palavra, o mapeador que executa as duas, e o envelope que sai. Tudo vem do exemplo, e os testes conferem que o guia diz o que o código faz.

1. O que a fonte devolve

Esta é a resposta do Livro para o processo LDG-2024-0001, gravada como a fonte a entrega. É o arquivo que os testes do exemplo usam:

examples/example_adapter/tests/payloads/ledger_proceeding.json

{
  "number": "LDG-2024-0001",
  "title": "Cobrança de aluguel",
  "class_name": "CIVEL COMUM",
  "secrecy": "RESERVADO",
  "filed_at": "2024-03-04T10:15:00",
  "claim_amount": "1500.50",
  "parties": [
    {"id": "p-1", "name": "Maria Souza", "side": "AUTOR", "kind": "PF"},
    {"id": "p-2", "name": "Comércio Alfa Ltda", "side": "REU", "kind": "PJ"},
    {"id": "p-3", "name": "Joaquim Lima", "side": "PERITO", "kind": "PF"}
  ],
  "events": [
    {"id": "e-1", "at": "2024-03-04T10:15:00", "code": "DIST", "label": "Distribuído"},
    {"id": "e-2", "at": "2024-03-05T09:00:00", "code": "JUNT", "label": "Juntada de petição"},
    {"id": "e-3", "at": "2024-03-06T14:30:00", "code": "VIST", "label": "Vistoria realizada"}
  ],
  "filings": [
    {
      "id": "f-1", "kind": "PET", "title": "Petição inicial", "event": "e-1",
      "created_at": "2024-03-04T10:15:00", "mimetype": "application/pdf", "size": 16
    },
    {"id": "f-2", "kind": "VIST", "title": "Termo de vistoria", "event": "e-3"},
    {"id": "f-3", "kind": "XYZ", "title": "Documento de outro tipo", "event": "e-9"}
  ]
}

Repare em cinco coisas, que o resto da seção explica: as datas vêm sem fuso; o sigilo vem como a palavra RESERVADO; o perito PERITO não tem equivalente nacional; a peça f-3 tem um tipo (XYZ) que o vocabulário não conhece; e a mesma peça aponta para um evento, e-9, que a fonte não mandou.

2. Campo a campo

A tabela diz, para cada campo da resposta, onde ele vai no envelope e o que o mapeador faz. Um caminho como parties[].side é o campo side de cada item de parties. O que o Livro não manda (o órgão julgador, a data de autuação, os assuntos) fica de fora do envelope: não se inventa.

Campo da fonte	Campo do envelope	O que o mapeador faz
number	proceeding.source_number e capture.source_number	Copia, sem mexer
title	proceeding.title	Copia
class_name	proceeding.procedural_class	Traduz: manda a palavra e o conceito
secrecy	proceeding.confidentiality e proceeding.confidentiality_rank	Traduz, e lê o nível da escala no vocabulário. Palavra desconhecida vai sem conceito e sem nível
filed_at	proceeding.filed_at	Aplica o fuso de SOURCE_TIMEZONE, porque a fonte grava sem fuso
claim_amount	proceeding.claim_value	Manda o número como texto decimal. Ausente ou zero fica de fora, porque zero afirmaria o que a fonte não afirmou
parties[].id	parties[].ref	Copia: é a referência da parte dentro do processo
parties[].name	parties[].name	Copia. A parte sem nome e sem documento torna a resposta inválida
parties[].side	parties[].role	Traduz: polo ativo ou polo passivo
parties[].kind	parties[].person_kind	Traduz: pessoa física ou jurídica
events[].id	events[].ref	Copia
events[].at	events[].occurred_at	Aplica o fuso
events[].code	events[].movements	Traduz o código. Só entra o movimento que o vocabulário conhece
events[].label	events[].source_label	Copia sempre: é o texto que a tela mostra quando não há tradução
filings[].id	filings[].source_identifier	Copia: é o identificador que o trabalho de peça traz depois, para buscar o arquivo
filings[].kind	filings[].type	Traduz pelo código. O tipo próprio do exemplo (VIST) sai com a chave que o adaptador declarou
filings[].title	filings[].label	Copia
filings[].event	filings[].event_refs	Vira a lista com esse evento, se a fonte o mandou; senão, lista vazia
filings[].created_at	filings[].created_at	Aplica o fuso
filings[].mimetype	filings[].declared_mimetype	Copia
filings[].size	filings[].size_bytes	Copia

3. Palavra a palavra

Onde o mapeador traduz, ele consulta o vocabulário, que é dado e não código. A tabela mostra cada palavra da resposta, a lista a que pertence e o conceito que sai. A palavra original vai sempre, e o conceito só quando o vocabulário o conhece.

Vocabulário	Palavra da fonte	Conceito enviado
confidentiality	label: RESERVADO	segredo_de_justica
procedural_class	label: CIVEL COMUM	7
role	label: AUTOR	polo_ativo
role	label: REU	polo_passivo
role	label: PERITO	sem tradução
person_kind	label: PF	fisica
person_kind	label: PJ	juridica
movement	code: DIST	26
movement	code: JUNT	67
movement	code: VIST	sem tradução
filing_type	code: PET	37
filing_type	code: VIST	termo-de-vistoria
filing_type	code: XYZ	sem tradução

O mesmo código VIST aparece duas vezes e vale duas coisas. Como código de evento, não tem tradução; como código de peça, é o tipo próprio que o adaptador declarou. O vocabulário é por fonte, por lista e por palavra, então a mesma palavra pode ter destinos diferentes em listas diferentes. O sigilo RESERVADO ainda leva o nível 1 da escala, que o vocabulário guarda ao lado do conceito.

4. O mapeador

O mapeador é uma função pura: lê a resposta gravada, consulta o vocabulário e devolve o envelope. A função principal junta o que cada parte do processo traduz:

examples/example_adapter/src/example_adapter/sources/ledger/mapper.py

    terms = _Terms(vocabulary, context)
    proceeding = compact(
        {
            "title": source.title,
            "procedural_class": terms("procedural_class", label=source.class_name),
            "confidentiality": terms("confidentiality", label=source.secrecy),
            "confidentiality_rank": vocabulary.rank(
                context.source_key, "confidentiality", label=source.secrecy
            ),
            "filed_at": instant(source.filed_at, context),
            "claim_value": _claim_value(source.claim_amount),
            "parties": [_party(party, terms) for party in source.parties],
            "events": [_event(event, terms, context) for event in source.events],
            "filings": [_filing(filing, source, terms, context) for filing in source.filings],
        }
    )
    return envelope(payload, source.number, proceeding, context, version)

O evento mostra duas regras. A palavra do evento vai sempre em source_label, e só entra em movements o movimento que o vocabulário traduziu:

examples/example_adapter/src/example_adapter/sources/ledger/mapper.py

def _event(event: _Event, terms: _Terms, context: MapContext) -> Json:
    movement = terms("movement", code=event.code, label=event.label)
    return {
        "ref": event.id,
        "occurred_at": instant(event.at, context),
        "source_label": event.label,
        # Só o movimento que o vocabulário traduziu: sem conceito, a palavra fica no rótulo.
        "movements": [movement] if movement and "concept" in movement else [],
    }

A peça mostra a terceira: a referência a um evento só vale se a fonte mandou esse evento.

examples/example_adapter/src/example_adapter/sources/ledger/mapper.py

def _filing(filing: _Filing, source: _Proceeding, terms: _Terms, context: MapContext) -> Json:
    known = {event.id for event in source.events}
    return compact(
        {
            "source_identifier": filing.id,
            "type": terms("filing_type", code=filing.kind),
            "label": filing.title,
            "created_at": instant(filing.created_at, context),
            "declared_mimetype": filing.mimetype,
            "size_bytes": filing.size,
            "event_refs": [filing.event] if filing.event in known else [],
        }
    )

O que envolve tudo é o envelope, que guarda a resposta da fonte como ela chegou, em base64, com o digest calculado sobre os mesmos bytes:

examples/example_adapter/src/example_adapter/envelope.py

def envelope(
    response: SourceResponse, number: str, proceeding: Json, context: MapContext, version: str
) -> Json:
    """O envelope do contrato: o bruto literal (comparável byte a byte) e o processo traduzido."""
    return {
        "contract_version": CONTRACT_VERSION,
        "adapter_version": version,
        "source_system": context.source_key,
        "job_id": context.job_id,
        "capture": {
            "source_number": number,
            "endpoint": response.endpoint,
            "content_type": "application/json",
            "content": base64.b64encode(response.body).decode(),
            "digest": f"sha256:{hashlib.sha256(response.body).hexdigest()}",
        },
        "proceeding": {"source_number": number, **proceeding},
    }

5. O envelope que sai

Este é o envelope que o mapeador devolve para a resposta gravada do passo 1, com o número do trabalho 42 e a versão 0.1.0 do adaptador. Está em examples/example_adapter/tests/expected/ledger_proceeding.envelope.json, e o teste test_the_envelope_the_guide_shows_is_what_the_mapper_returns_for_the_recorded_payload garante que o mapeador devolve exatamente isto. O começo traz a identificação e a captura (o content é a resposta inteira em base64, que o guia omite):

examples/example_adapter/tests/expected/ledger_proceeding.envelope.json

{
  "contract_version": "2",
  "adapter_version": "0.1.0",
  "source_system": "ledger",
  "job_id": 42,
  "capture": {
    "source_number": "LDG-2024-0001",
    "endpoint": "/proceedings/LDG-2024-0001",
    "content_type": "application/json",
...
    "digest": "sha256:5da61095e45745947e63f1dd17331707bb97ddd86b2b2a5d6f3523f5b6426b78"
  },

Na sequência, o processo: a classe e o sigilo viram termos com a palavra original e o conceito, o sigilo leva também o nível, a data ganha o fuso, e o valor da causa vai como texto decimal.

examples/example_adapter/tests/expected/ledger_proceeding.envelope.json

  "proceeding": {
    "source_number": "LDG-2024-0001",
    "title": "Cobrança de aluguel",
    "procedural_class": {
      "vocabulary": "procedural_class",
      "label": "CIVEL COMUM",
      "concept": "7"
    },
    "confidentiality": {
      "vocabulary": "confidentiality",
      "label": "RESERVADO",
      "concept": "segredo_de_justica"
    },
    "confidentiality_rank": 1,
    "filed_at": "2024-03-04T10:15:00-03:00",
    "claim_value": "1500.50",

O perito PERITO vai só com a palavra, sem conceito. A tela do produto o mostra como a fonte o escreveu, com a marca "sem tradução":

examples/example_adapter/tests/expected/ledger_proceeding.envelope.json

      {
        "ref": "p-3",
        "role": {
          "vocabulary": "role",
          "label": "PERITO"
        },
        "name": "Joaquim Lima",
        "person_kind": {
          "vocabulary": "person_kind",
          "label": "PF",
          "concept": "fisica"
        }
      }

O evento e-3, de um movimento sem tradução, vai com a palavra em source_label e sem movimento:

examples/example_adapter/tests/expected/ledger_proceeding.envelope.json

      {
        "ref": "e-3",
        "occurred_at": "2024-03-06T14:30:00-03:00",
        "source_label": "Vistoria realizada",
        "movements": []
      }

E a peça f-3, de tipo desconhecido e que citava um evento inexistente, vai sem conceito e sem referência a evento:

examples/example_adapter/tests/expected/ledger_proceeding.envelope.json

      {
        "source_identifier": "f-3",
        "type": {
          "vocabulary": "filing_type",
          "code": "XYZ"
        },
        "label": "Documento de outro tipo",
        "event_refs": []
      }

Nada neste envelope foi inventado: o que a fonte não disse não aparece.

Fonte sem padrão

Ao fim desta seção você tem o roteiro para escrever o adaptador de uma fonte que não segue padrão nenhum, como a API privada de um sistema da instituição, sem documentação, ou com uma documentação que não confere com o que a API responde.

Fonte com padrão público traz o formato pronto, e você só o traduz. Fonte sem padrão pede antes que você descubra o formato, e a única verdade é o que ela responde. O Livro e o Arquivo do exemplo são assim: o formato deles é o que as respostas gravadas mostram, e nada além. O processo de entrada de uma instituição, do primeiro contato à instalação, está em apps/docs/content/integrar/playbook-novo-cliente.md. O trabalho de código vem nesta ordem:

1. Reconheça a API. Descubra quais rotas devolvem o que o produto precisa: o processo com partes, eventos e peças, o arquivo de cada peça e a busca por número. As fontes de informação são a documentação, se existir, a pessoa da instituição que mantém o sistema e as chamadas que a própria tela da instituição faz (as ferramentas do desenvolvedor do navegador mostram cada chamada). Peça uma credencial só de leitura. Anote como a API autentica, como ela diz "não existe" (o status), como diz "agora não posso" e se ela pagina. Comece com um RATE_LIMIT_PER_SECOND baixo: ninguém sabe ainda o limite dela.
2. Grave respostas reais. Para cada caso, grave em arquivo o corpo exato que a fonte devolve, um arquivo por caso, em tests/payloads/: o processo completo, o mínimo, o que não existe, o que a fonte recusa, o corpo ilegível e a pausa por limite. Grave os bytes sem reformatar, porque o Backend guarda a captura e o digest dela, calculado sobre esses bytes. Resposta real tem dado pessoal: antes de gravar no repositório, troque nomes, documentos e números por dados fictícios que mantenham a forma (os mesmos campos, os mesmos tipos, as mesmas palavras). Resposta real sem essa troca nunca entra no git.
3. Escreva a tabela de mapeamento e entregue-a à instituição. Uma linha para cada campo das respostas gravadas, no modelo abaixo (o Livro de Registro, na seção anterior, tem a tabela completa). Quem confere é a instituição: ela confirma o que cada campo e cada palavra significam, e você não adivinha. A última coluna guarda quem confirmou e quando.
4. Escreva o cliente. Só E/S, pelo SourceHttp do kit. Veja "Classificar as falhas" e "Fonte que limita o ritmo ou falha". Teste com um httpx.MockTransport que serve as respostas gravadas.
5. Escreva o mapeador. Uma função pura que segue a tabela, testada contra as respostas gravadas. Cada teste afirma o que o contrato e as regras do produto pedem do envelope, não o que o código faz. Comece pelo obrigatório, como mostra a próxima parte.
6. Escreva o vocabulário. Cada palavra confirmada pela instituição vira uma entrada do terms.json. A palavra que ninguém confirmou não entra, e vai sem conceito. O teste que confere as chaves contra as publicadas pega o erro de digitação.
7. Prove com a suíte de conformidade. Rode o adaptador inteiro com o run_adapter, uma fonte simulada que serve as respostas gravadas e um trabalho por caso. Se a suíte passa, o adaptador fala o contrato, e falta só ligá-lo à fonte de verdade, com o ritmo baixo, e medir.

O modelo da tabela de mapeamento:

Campo da fonte	Amostra gravada	Campo do envelope	Regra de tradução	Confirmado por
secrecy	RESERVADO	confidentiality e confidentiality_rank	RESERVADO é segredo de justiça, nível 1	nome e data
claim_amount	1500.50	claim_value	Texto decimal; zero e ausente ficam de fora	nome e data
side	PERITO	role	Sem conceito nacional: vai só com a palavra	nome e data

Comece pelo obrigatório

A primeira versão do mapeador manda só o que o contrato exige: o número do processo e a captura. Isso já prova o caminho inteiro, do cliente ao Backend, e deixa a instituição ver o primeiro feito chegar. Depois você enriquece, um grupo de campos por vez, cada um com a linha da tabela confirmada e o seu teste: título e classe, sigilo, datas, partes, eventos e peças. Ligue o sigilo logo no começo, porque sem ele o feito entra no nível máximo e quase ninguém o vê.

O exemplo tem esse caso. A fonte devolve o número e um valor da causa zero, que na fonte quer dizer "não informado":

examples/example_adapter/tests/payloads/ledger_minimal.json

{"number": "LDG-2024-0002", "claim_amount": "0"}

E o teste afirma o envelope mínimo, em que as listas vazias são o padrão do contrato:

examples/example_adapter/tests/test_example_mappers.py

    # Valor da causa zero é "a fonte não disse", não "zero".
    assert proceeding == {
        "source_number": "LDG-2024-0002",
        "parties": [],
        "events": [],
        "filings": [],
    }

Nunca invente, e lembre que o sigilo falha fechado

• Campo que a fonte não manda fica de fora, nunca com valor de enchimento. O valor da causa que falta é ausente, não 0. Uma competência que a fonte não tem é ausente. Um evento sem data é inválido, e o mapeador levanta InvalidPayloadError; não existe data zero.
• Palavra que o vocabulário não conhece vai sem conceito. Nunca escolha um conceito parecido só para a suíte passar: o Backend recusa o envelope inteiro se o conceito não está na lista publicada.
• Sem nível de sigilo, o Backend aplica o máximo. Nunca mande "público" por falta do dado: o nível vem da fonte, ou não vem.

Casos de uso

Cada seção abaixo ensina um caso que uma equipe encontra, com o código do exemplo que o resolve.

Uma segunda fonte no mesmo adaptador

Ao fim você registra mais uma fonte, e o kit roteia cada trabalho para ela pela chave.

Uma instituição costuma ter mais de uma fonte, ou mais de uma instância da mesma. O kit atende todas com o mesmo laço: cada trabalho da fila traz a chave da fonte (source_system), e o kit o entrega ao cliente e ao mapeador registrados para ela. São três passos.

1. Declare a fonte em declaration.py. A chave segue [a-z0-9_-]+, e a ordem da lista é a ordem em que a navegação mostra as fontes:

examples/example_adapter/src/example_adapter/declaration.py

# Na ordem em que a navegação as mostra.
SOURCES = (
    SourceSystemDeclaration(
        key="ledger",
        protocol="ledger-rest",
        version="1.0",
        short_label="Livro",
        label="Livro de Registro",
        requestable=True,
    ),
    SourceSystemDeclaration(
        key="archive",
        protocol="archive-rest",
        version="1.0",
        short_label="Arquivo",
        label="Arquivo Central",
        requestable=False,
    ),
)

2. Escreva o cliente e o mapeador dela, em sources/<sistema>/.
3. Registre uma fonte por chave declarada na montagem:

examples/example_adapter/src/example_adapter/__main__.py

    return Adapter(
        declaration(),
        {
            "ledger": Source(
                LedgerClient(settings, source_http),
                partial(map_ledger, vocabulary=vocabulary, version=ADAPTER_VERSION),
            ),
            "archive": Source(
                ArchiveClient(settings, source_http, ArchiveLogin(settings, http)),
                partial(map_archive, vocabulary=vocabulary, version=ADAPTER_VERSION),
            ),
        },
    )

Duas regras valem. Uma chave declarada sem fonte registrada deixa trabalho sem quem o atenda; o exemplo tem um teste que falha se as chaves da declaração e as do registro divergirem. E um trabalho cuja chave não tem fonte registrada vira falha invalid, nomeando a chave, em vez de travar a fila.

Duas chaves podem dividir um cliente, quando são duas instâncias do mesmo sistema (dois graus de um tribunal, por exemplo). Passe a mesma instância do cliente às duas fontes: elas dividem também a vazão e a pausa por limite, porque a fonte conta tudo junto. O adaptador real do repositório, em adapters/mprs/, faz assim.

Tipos de peça próprios

Ao fim você manda ao Backend um tipo de peça que a lista nacional não tem.

O Backend só aceita no envelope os tipos de peça da lista nacional e os que o adaptador declarou. Se a fonte tem um tipo sem equivalente nacional, declare-o, e depois traduza a palavra da fonte para a chave dele. Na declaração:

examples/example_adapter/src/example_adapter/declaration.py

# Tipos de peça que a lista nacional não tem. O vocabulário só pode mandar estas chaves.
OWN_FILING_TYPES = (FilingTypeDeclaration(key="termo-de-vistoria", label="Termo de vistoria"),)


def declaration() -> Declaration:
    return Declaration(
        contract_version=CONTRACT_VERSION,
        systems=list(SOURCES),
        adapter_version=ADAPTER_VERSION,
        filing_types=list(OWN_FILING_TYPES),
    )

E no vocabulário, a palavra da fonte aponta para a chave declarada:

examples/example_adapter/src/example_adapter/vocabulary/terms.json

    {"vocabulary": "filing_type", "concept": "termo-de-vistoria", "source_terms": [
      {"source_system": "ledger", "code": "VIST"}
    ]}

Cuidados que custaram caro a quem veio antes:

• O tipo próprio só vale em filing_type. A mesma chave em outro vocabulário é recusada, nomeando vocabulário:chave.
• Vale a última declaração aceita. Um tipo que uma declaração nova deixa de fora para de valer.
• Declaração recusada pelo Backend (422) não muda nada: os tipos da última aceita continuam valendo. O kit registra a recusa e insiste.
• O exemplo confere que todo conceito do vocabulário é uma chave publicada ou um tipo declarado, antes de qualquer envio (test_example_shape.py).

Mapear uma palavra da fonte para um conceito, com a palavra original

Ao fim você traduz um valor da fonte para o conceito nacional sem perder o que a fonte escreveu.

Valor que veio da fonte nunca vira coluna crua, nem sendo um número, uma sigla ou um verdadeiro ou falso. O mapeador o traduz para um conceito nacional e manda o conceito junto com a palavra original, que fica como registro. Essa dupla é o source_term do contrato: vocabulary (qual lista), code e label (a palavra da fonte) e concept (a chave nacional). Duas situações:

A fonte diz	O mapeador envia
secrecy: "RESERVADO", que o vocabulário conhece	{"vocabulary": "confidentiality", "label": "RESERVADO", "concept": "segredo_de_justica"}, mais o nível 1 da escala
side: "PERITO", que o vocabulário não conhece	{"vocabulary": "role", "label": "PERITO"}, sem concept

O vocabulário é dado, não código. Cada entrada diz o conceito e as palavras de cada fonte que apontam para ele, e o nível de sigilo quando o conceito é de sigilo:

examples/example_adapter/src/example_adapter/vocabulary/terms.json

    {"vocabulary": "confidentiality", "concept": "publico", "rank": 0, "source_terms": [
      {"source_system": "ledger", "label": "ABERTO"},
      {"source_system": "archive", "label": "PUBLICO"}
    ]},
    {"vocabulary": "confidentiality", "concept": "segredo_de_justica", "rank": 1, "source_terms": [
      {"source_system": "ledger", "label": "RESERVADO"},
      {"source_system": "archive", "label": "RESTRITO"}
    ]},

A consulta devolve sempre a palavra original, e o conceito só quando o vocabulário o conhece:

examples/example_adapter/src/example_adapter/vocabulary/__init__.py

    def term(
        self,
        source_system: str,
        vocabulary: str,
        *,
        code: str | None = None,
        label: str | None = None,
    ) -> Json | None:
        """O termo do contrato: a palavra original sempre, o conceito só se o vocabulário o
        conhece. Sem palavra, sem termo, e a string vazia não é palavra."""
        code, label = code or None, label or None
        if code is None and label is None:
            return None
        found = self._find(source_system, vocabulary, code, label)
        term: Json = {"vocabulary": vocabulary}
        for key, value in (
            ("code", code),
            ("label", label),
            ("concept", found.concept if found else None),
        ):
            if value is not None:
                term[key] = value
        return term

O mapeador usa a consulta para cada campo traduzido, com a chave da fonte vinda do contexto do trabalho, nunca escrita à mão:

examples/example_adapter/src/example_adapter/sources/ledger/mapper.py

def _party(party: _Party, terms: _Terms) -> Json:
    # O Backend recusa a parte sem nome e sem documento, e a string vazia conta como ausente.
    if not (party.name or party.document):
        raise InvalidPayloadError(f"parte {party.id} sem nome e sem documento")
    return compact(
        {
            "ref": party.id,
            "role": terms("role", label=party.side),
            "name": party.name or None,
            "document_number": party.document or None,
            "person_kind": terms("person_kind", label=party.kind),
        }
    )

Nunca invente um conceito

Palavra que o vocabulário não conhece vai sem conceito, e a tela do produto a mostra como a fonte a escreveu, com a marca "sem tradução". Nunca invente um conceito para a palavra passar: o Backend recusa o envelope inteiro se o conceito não está na lista publicada, e o feito espera.

A parte sem nome e sem documento o mapeador recusa: levanta InvalidPayloadError, e o trabalho falha como invalid. Mandá-la só com ref e polo seria recusado pelo Backend (party-name-or-document), e um nome inventado seria pior. A string vazia também não é palavra: a consulta a trata como ausente, e o termo sem código e sem rótulo nem sai do mapeador.

Datas: a fonte costuma gravar a data sem fuso. O kit lê o fuso de SOURCE_TIMEZONE e o entrega ao mapeador em context.timezone; o mapeador o aplica uma vez só:

examples/example_adapter/src/example_adapter/envelope.py

def instant(moment: datetime | None, context: MapContext) -> str | None:
    """A fonte grava a data sem offset: vale o fuso declarado em `SOURCE_TIMEZONE`."""
    if moment is None:
        return None
    return (moment if moment.tzinfo else moment.replace(tzinfo=context.timezone)).isoformat()

Onde estão as chaves: as chaves de conceito que o envelope pode citar estão em src/proceedings_adapter_kit/contract/schemas/concept-keys.json, e a instalação serve as suas em GET /work/contract/concept-keys, com o token da ingestão.

Entregar o arquivo de uma peça, por ticket

Ao fim você entrega o arquivo de uma peça, sem escrever uma linha do envio e sem nunca o ter inteiro em memória.

Alguns trabalhos da fila pedem o arquivo de uma peça, não o feito: o trabalho vem com kind: "filing", o identificador da peça no Backend (filing_id) e o da fonte (filing_source_identifier). O arquivo não passa pelo Backend. O kit pede um ticket, que é uma sessão de envio retomável do armazenamento de objetos, manda a ela o fluxo de bytes em partes de 8 MiB, à medida que a fonte o entrega, e avisa o Backend, que confere o objeto antes de a peça ficar disponível. O tamanho e o tipo do conteúdo só se sabem ao fim do fluxo, por isso vão na conclusão, e não no pedido do ticket.

O que você escreve é só a busca do arquivo na fonte: devolva o fluxo de bytes já decodificados (chunks) e uma função que devolve o tipo (mimetype). O kit chama essa função depois de ler o fluxo, porque o corpo da fonte pode trazer o tipo no fim. Abra a conexão com a fonte dentro do gerador, como o exemplo faz: o status que a fonte recusa sai na primeira leitura, e uma entrega que falha antes de ler não deixa conexão aberta.

examples/example_adapter/src/example_adapter/sources/ledger/client.py

    async def fetch_filing(
        self, source_key: str, number: str, filing_source_identifier: str
    ) -> FilingFile:
        """Mesma credencial e mesma vazão do processo: a fonte conta as duas coisas juntas. O corpo
        vai ao kit em blocos, e a conexão se abre dentro do gerador: o status recusado sai na
        primeira leitura, e uma entrega que falha antes de ler não deixa conexão aberta."""
        url = (
            f"{self._base_url}/proceedings/{quote(number, safe='')}"
            f"/filings/{quote(filing_source_identifier, safe='')}"
        )
        seen: dict[str, str | None] = {}

        async def chunks() -> AsyncIterator[bytes]:
            async with self._http.stream(
                url,
                headers=self._headers,
                not_found_message=f"peça não encontrada na fonte: {filing_source_identifier}",
            ) as response:
                seen["type"] = response.headers.get("content-type")
                async for block in response.aiter_bytes():
                    yield block

        return FilingFile(chunks=chunks(), mimetype=lambda: seen.get("type"))

FilingFile é o que o kit espera de volta:

src/proceedings_adapter_kit/source/protocols.py

@dataclass(frozen=True, slots=True)
class FilingFile:
    """O arquivo de uma peça, como a fonte o entrega: um fluxo de bytes já decodificados, que o kit
    manda ao bucket à medida que chegam, sem nunca o ter inteiro em memória. `mimetype` se lê
    depois do fluxo, porque o corpo da fonte pode trazer o tipo no fim. O erro da fonte no meio do
    fluxo sai como o de qualquer busca (`TransientSourceError`, `SourceRejectedError`)."""

    chunks: AsyncIterator[bytes]
    mimetype: Callable[[], str | None]

Regras do envio, todas conferidas pelo Backend falso:

• Cada parte do meio tem um múltiplo de 256 KiB (o kit usa 8 MiB), e só a última é menor. Uma parte do meio de outro tamanho é recusada.
• Cada parte começa no byte seguinte ao último que a sessão guarda: sem lacuna e sem repetição. A parte que falha é reenviada desse ponto, até 3 vezes.
• O total só vai na última parte, e é o dos bytes enviados. Quando o fluxo termina na borda de uma parte, o fim é um PUT sem corpo com o total, e o arquivo vazio fecha com total zero.
• A parte leva os cabeçalhos do ticket e nenhum Authorization: a URL é a credencial, e o token da ingestão nunca vai ao armazenamento.
• A sessão fechada recusa outra parte com 412, porque o objeto já existe.
• A conclusão leva o tamanho que o kit contou, e o Backend o confere no objeto. Outro tamanho é recusado.
• O erro que a fonte levanta no meio do fluxo (TransientSourceError, SourceRejectedError) sai como o de qualquer busca, e nada se conclui.
• Qualquer passo que falhe (ticket, envio ou conclusão) vira falha transient, e a fila repete a peça inteira com um ticket novo.

Se a fonte não serve arquivos, recuse o pedido de peça, em vez de deixar o trabalho preso. O Arquivo Central do exemplo só guarda o histórico:

examples/example_adapter/src/example_adapter/sources/archive/client.py

    async def fetch_filing(
        self, source_key: str, number: str, filing_source_identifier: str
    ) -> FilingFile:
        """O Arquivo guarda só o histórico: peça pedida a ele é recusa, e repetir não conserta."""
        raise SourceRejectedError("o Arquivo Central não serve peças")

Classificar as falhas

Ao fim você sabe que falha levantar em cada situação, e que relato a fila recebe.

A fila precisa saber se vale repetir. Por isso o kit traduz cada falha em um de três relatos: transient (vale repetir), not_found (repetir não muda a resposta) e invalid (repetir não conserta). Quem classifica é o SourceHttp do kit, para o que é HTTP, e o seu código, para o que é leitura da resposta.

Situação	Quem levanta	Relato
A fonte diz que não existe, no status que você declara em not_found (padrão 404)	SourceHttp	not_found
429, ou 503 com Retry-After	SourceHttp	transient, com pausa
5xx, 401, 403, conexão derrubada, tempo esgotado	SourceHttp	transient
Outro 4xx	SourceHttp, como SourceRejectedError	invalid
Corpo que o mapeador não consegue ler	Seu mapeador, com InvalidPayloadError	invalid
Login da fonte indisponível	Seu código, com TransientSourceError	transient
O Backend recusa o envelope (422)	Kit, como ContractRejectedError	invalid
Falha no envio da peça (ticket, envio ou conclusão)	Kit, como FilingDeliveryError	transient
Defeito do seu código com um corpo específico	Exceção qualquer	transient, com o registro inteiro do defeito
O Backend fora do ar	Kit, como BackendError	Nenhum relato: o trabalho volta à fila sozinho

O laço do kit tem exatamente esta tabela:

src/proceedings_adapter_kit/worker/loop.py

            except SourceNotFoundError as error:
                await self._report(job, "not_found", error, started)
            except (TransientSourceError, FilingDeliveryError) as error:
                await self._report(job, "transient", error, started)
            except PERMANENT as error:
                await self._report(job, "invalid", error, started)
            except BackendError as error:
                log.warning("ingest_unreachable", job_id=job.job_id, error=str(error))
            except Exception as error:
                # Defeito do adaptador com este payload: registra inteiro e deixa a fila
                # repetir até o limite, em vez de derrubar os outros jobs do lote.
                log.exception("job_crashed", job_id=job.job_id)
                await self._report(job, "transient", error, started)

Cada trabalho entregue ou relatado termina numa linha do registro, job_ingested ou job_failed, com stage (adapter), outcome (ok, ou o relato da tabela) e seconds, o tempo da tomada do trabalho ao relato. Os mesmos campos estão nas linhas das outras etapas da plataforma, e é deles que sai o painel de cada uma.

Do seu lado, são dois lugares. No cliente, diga ao SourceHttp com que status a fonte responde "não existe". O Arquivo Central responde 410, e o 404 dele, por isso, vira recusa, não "não encontrado":

examples/example_adapter/src/example_adapter/sources/archive/client.py

    async def fetch_proceeding(self, source_key: str, number: str) -> SourceResponse:
        endpoint = f"/dossiers/{quote(number, safe='')}"
        response = await self._http.get(
            f"{self._base_url}{endpoint}",
            headers={"authorization": f"Bearer {await self._login.token()}"},
            not_found=NOT_FOUND,
            not_found_message=f"dossiê não encontrado na fonte: {number}",
        )
        return SourceResponse(endpoint=endpoint, body=response.content)

E no mapeador, levante InvalidPayloadError para o que não dá para ler, sem inventar valor para o campo que faltou:

examples/example_adapter/src/example_adapter/sources/ledger/mapper.py

def map_ledger(
    payload: object, context: MapContext, *, vocabulary: Vocabulary, version: str
) -> Json:
    if not isinstance(payload, SourceResponse):
        raise InvalidPayloadError(f"payload não é do cliente do Livro: {type(payload).__name__}")
    try:
        source = _Proceeding.model_validate_json(payload.body)
    except ValidationError as error:
        raise InvalidPayloadError(
            f"payload fora do formato do Livro: {error.error_count()} erro(s)"
        ) from error

O exemplo prova um caso de cada tipo contra o Backend falso: não encontrado, falha transitória, recusa e corpo ilegível, cada um no relato certo (test_example_conformance.py).

Fonte que limita o ritmo ou falha

Ao fim você sabe como o kit protege a fonte de você, e como ajustar o ritmo.

Você não escreve controle de ritmo. Basta chamar sempre SourceHttp.get, e nunca o httpx direto, que passaria por cima do controle. O SourceHttp faz três coisas por cliente, e vale para todos os pedidos concorrentes dele, de feito e de peça, porque a fonte conta tudo junto:

• Ritmo adaptativo: sai um pedido a cada 1 / taxa segundos, sem rajada. A taxa começa no teto (RATE_LIMIT_PER_SECOND), cai à metade, no máximo uma vez por segundo, quando a fonte se sobrecarrega (limite, conexão recusada ou expirada), e sobe um pedido por segundo a cada 20 sucessos seguidos, até o teto. É o controle do TCP.
• Pausa por Retry-After: quando a fonte responde 429, ou 503 com Retry-After, nenhum pedido sai até ela liberar. O valor pode vir em segundos ou como data HTTP. Sem o cabeçalho, a pausa começa em 5 segundos, dobra a cada limite seguido até 300, e volta ao começo no primeiro sucesso.
• Medidas: a cada 10 segundos o kit registra a vazão: taxa, pedidos por segundo, sobrecargas e a latência (p50 e p95), uma vez por cliente.

Do kit, o trecho que decide a pausa:

src/proceedings_adapter_kit/source/throttle.py

        asked = _seconds(retry_after)
        if asked is None:
            delay = self._next
            self._next = min(self._next * 2, self.MAX_SECONDS)
        else:
            delay = asked
        self._until = max(self._until, time.monotonic() + delay)
        return delay

No cliente, a única coisa que você faz é passar o pedido pelo SourceHttp:

examples/example_adapter/src/example_adapter/sources/ledger/client.py

    async def fetch_proceeding(self, source_key: str, number: str) -> SourceResponse:
        endpoint = f"/proceedings/{quote(number, safe='')}"
        response = await self._http.get(
            f"{self._base_url}{endpoint}",
            headers=self._headers,
            not_found_message=f"processo não encontrado na fonte: {number}",
        )
        return SourceResponse(endpoint=endpoint, body=response.content)

O exemplo prova a pausa: uma fonte responde 429 com Retry-After: 1, o trabalho é relatado como transient, e o pedido seguinte à mesma fonte só sai depois do que ela pediu (test_a_429_with_retry_after_pauses_every_request_of_the_source).

Se a fonte é mais frágil do que o ritmo pede, baixe RATE_LIMIT_PER_SECOND e CONCURRENCY na instalação, sem tocar no código. Para escolher os valores, meça: o registro de vazão mostra se o ritmo chega perto da quota e por que não.

Fonte com login próprio

Ao fim você autentica na fonte com um token que o adaptador pede e renova sozinho.

Quando a fonte exige login, o login é E/S, e mora ao lado do cliente. O do exemplo pede um token de aplicação com credenciais de cliente, guarda o token e o renova perto de vencer. O trinco impede que buscas concorrentes peçam o token juntas:

examples/example_adapter/src/example_adapter/sources/archive/login.py

    async def token(self) -> str:
        async with self._lock:
            if self._token is None or self._expires_at - time.monotonic() < RENEW_MARGIN_SECONDS:
                self._token, self._expires_at = await self._fetch()
            return self._token

Login que falha é falha transient: sem token não há busca, mas repetir pode resolver.

examples/example_adapter/src/example_adapter/sources/archive/login.py

        except httpx.TransportError as error:
            raise TransientSourceError(f"login: {type(error).__name__}") from error
        # Sem token não há busca, mas repetir pode resolver: é falha transitória.
        if response.status_code != httpx.codes.OK:
            raise TransientSourceError(f"login: HTTP {response.status_code}")

O cliente pede o token a cada busca (que sai da memória, na maioria das vezes) e o põe no cabeçalho:

examples/example_adapter/src/example_adapter/sources/archive/client.py

    async def fetch_proceeding(self, source_key: str, number: str) -> SourceResponse:
        endpoint = f"/dossiers/{quote(number, safe='')}"
        response = await self._http.get(
            f"{self._base_url}{endpoint}",
            headers={"authorization": f"Bearer {await self._login.token()}"},
            not_found=NOT_FOUND,
            not_found_message=f"dossiê não encontrado na fonte: {number}",
        )
        return SourceResponse(endpoint=endpoint, body=response.content)

Na montagem, o login usa o cliente HTTP comum, e as buscas usam o cliente com a conexão viva por vaga que o kit monta (source_http_client):

examples/example_adapter/src/example_adapter/__main__.py

            "archive": Source(
                ArchiveClient(settings, source_http, ArchiveLogin(settings, http)),
                partial(map_archive, vocabulary=vocabulary, version=ADAPTER_VERSION),
            ),

Credencial não vai para mensagem de erro

Guarde as credenciais como SecretStr nas configurações e nunca as ponha em mensagem de erro: o exemplo registra só o tipo da falha de rede e o status HTTP.

As regras que quebram o produto se ignoradas

Ao fim você sabe o que o produto não tolera, e quem confere cada regra.

Estas regras vêm da experiência de quem escreveu o primeiro adaptador. A suíte de conformidade confere as que o contrato permite conferir, e o exemplo mostra como testar as que dependem do seu código. Onde nenhuma regra automática cobre, a tabela diz.

Regra	Por quê	Quem confere
Nenhum banco: o adaptador só fala HTTP	Quem escreve direto no banco passa por cima das regras do Backend	O kit não traz driver de banco (tests/test_kit_boundaries.py). No exemplo, test_example_shape.py recusa importação que não seja do kit, das bibliotecas que o kit instala ou da biblioteca padrão. A suíte de conformidade não vê isso
Toda mensagem no formato do contrato	O Backend recusa o que está fora do esquema	Regra schema, nomeando o campo
contract_version em toda mensagem	Mensagem sem a versão, ou com outra, é recusada com a versão que o Backend fala	Regra contract-version
Valor da fonte nunca vira coluna crua	O adaptador traduz para o conceito nacional e manda a palavra original junto	Regra concept-key: conceito fora das chaves publicadas e dos tipos declarados é recusado, nomeando vocabulário:chave
O envelope fecha por dentro	O Backend confere, além do esquema, o termo, a parte, o fuso, as referências e a captura	Regras term-code-or-label, party-name-or-document, timezone, reference, capture-base64 e capture-digest
Nunca inventar valor	Conceito que a fonte não manda fica nulo; valor da causa é nulo, não zero; evento sem data é inválido, não data zero	Nenhuma regra do Backend falso. O exemplo testa no mapeador (test_example_mappers.py)
Sigilo falha fechado	Sem o nível de sigilo, o Backend aplica o máximo. Nunca mande "público" por falta do dado	Nenhuma regra do Backend falso. O exemplo testa que palavra sem tradução não manda nível (test_example_mappers.py)
A peça só vai pelo ticket	O arquivo não passa pelo Backend, e a URL do ticket é a única porta para o objeto	Regras ticket-unknown, ticket-header-missing, ticket-header-changed, ticket-reused, content-range, chunk-alignment, chunk-offset, total-mismatch e completion-size
O token da ingestão nunca vai ao armazenamento de objetos	A URL do ticket é a credencial; mandar o token a mais vaza o segredo	Regra bucket-authorization
Uma fonte registrada por chave declarada	Trabalho de chave sem fonte não tem quem o atenda	O laço do kit relata invalid, nomeando a chave; o exemplo confere declaração e registro (test_example_shape.py)
O token do feed nunca entra na imagem por ARG nem ENV	Um token numa camada fica na imagem para sempre	O exemplo confere o Dockerfile (test_example_shape.py)

A suíte de conformidade nomeia estas regras (Rule), e a recusa carrega o nome:

src/proceedings_adapter_kit/testing/fake_backend.py

class Rule(StrEnum):
    """Cada regra que o backend falso aplica. É o nome que a recusa carrega."""

    SCHEMA = "schema"
    CONCEPT_KEY = "concept-key"
    CONTRACT_VERSION = "contract-version"
    TERM_CODE_OR_LABEL = "term-code-or-label"
    PARTY_NAME_OR_DOCUMENT = "party-name-or-document"
    TIMEZONE = "timezone"
    REFERENCE = "reference"
    CAPTURE_BASE64 = "capture-base64"
    CAPTURE_DIGEST = "capture-digest"
    TICKET_UNKNOWN = "ticket-unknown"
    TICKET_HEADER_MISSING = "ticket-header-missing"
    TICKET_HEADER_CHANGED = "ticket-header-changed"
    TICKET_REUSED = "ticket-reused"
    CONTENT_RANGE = "content-range"
    CHUNK_ALIGNMENT = "chunk-alignment"
    CHUNK_OFFSET = "chunk-offset"
    TOTAL_MISMATCH = "total-mismatch"
    COMPLETION_SIZE = "completion-size"
    BUCKET_AUTHORIZATION = "bucket-authorization"

Imagem e implantação

Ao fim você tem uma imagem do adaptador que lê o kit do feed sem deixar o token nela, e um compose da instituição que a sobe ao lado do produto.

A imagem fala só HTTP, com a fonte e com a porta de ingestão. O token do feed do kit entra só pelo segredo do BuildKit, montado no RUN: nunca por ARG nem por ENV, que o deixariam numa camada. Todo uv sync do Dockerfile o recebe assim:

examples/example_adapter/Dockerfile

RUN --mount=type=cache,target=/root/.cache/uv \
    --mount=type=secret,id=uv_index_xertica,env=UV_INDEX_XERTICA_PASSWORD \
    --mount=type=bind,source=uv.lock,target=uv.lock \
    --mount=type=bind,source=pyproject.toml,target=pyproject.toml \
    UV_INDEX_XERTICA_USERNAME=aad uv sync --frozen --no-install-project --no-editable

COPY pyproject.toml uv.lock ./
COPY src src
RUN --mount=type=cache,target=/root/.cache/uv \
    --mount=type=secret,id=uv_index_xertica,env=UV_INDEX_XERTICA_PASSWORD \
    UV_INDEX_XERTICA_USERNAME=aad uv sync --locked --no-editable

E a imagem final roda sem privilégio, com o ambiente pronto:

examples/example_adapter/Dockerfile

FROM python:3.14.7-slim-trixie
RUN groupadd --system --gid 999 app \
 && useradd --system --gid 999 --uid 999 --create-home app
COPY --from=builder --chown=app:app /app/.venv /app/.venv
ENV PATH="/app/.venv/bin:$PATH" PYTHONUNBUFFERED=1
WORKDIR /app
USER app
CMD ["python", "-m", "example_adapter"]

A construção da imagem parte da pasta do adaptador, com o token no ambiente (um token curto do login da Microsoft (Azure AD), ou um PAT com leitura de pacotes, como no começo rápido):

UV_INDEX_XERTICA_PASSWORD=... docker build --secret id=uv_index_xertica,env=UV_INDEX_XERTICA_PASSWORD adapters/<nome>

O .dockerignore do adaptador é uma contenção: ele segura o .env e o .venv fora do contexto de construção, e o exemplo tem um teste que confere (test_example_shape.py).

A instituição sobe o adaptador com um compose que se soma ao do produto. O adaptador declara ao Backend as fontes, a versão e os tipos de peça, e só pega trabalho depois que o Backend aceita, por isso espera o Backend saudável. Réplicas podem subir lado a lado: a fila entrega cada trabalho a uma réplica só.

examples/example_adapter/installation/compose.yaml

services:
  adapter:
    # O kit vem do feed da Xertica: o build lê o token do ambiente (`UV_INDEX_XERTICA_PASSWORD`)
    # por segredo do BuildKit, que não fica na imagem.
    build:
      context: adapters/example
      secrets: [uv_index_xertica]
    image: proceedings-example-adapter
    environment:
      BACKEND_URL: http://backend:8080
      # O mesmo segredo que o Backend lê em INGESTION_SERVICE_TOKEN, do .env (fora do git).
      INGESTION_SERVICE_TOKEN: ${INGESTION_SERVICE_TOKEN:?defina INGESTION_SERVICE_TOKEN no .env}
      LEDGER_BASE_URL: ${LEDGER_BASE_URL:-}
      LEDGER_API_KEY: ${LEDGER_API_KEY:-}
      ARCHIVE_BASE_URL: ${ARCHIVE_BASE_URL:-}
      ARCHIVE_TOKEN_URL: ${ARCHIVE_TOKEN_URL:-}
      ARCHIVE_CLIENT_ID: ${ARCHIVE_CLIENT_ID:-}
      ARCHIVE_CLIENT_SECRET: ${ARCHIVE_CLIENT_SECRET:-}
      # As fontes não publicam limite e nada foi medido: comece conservador e meça.
      RATE_LIMIT_PER_SECOND: "5"
      CONCURRENCY: "8"
      SOURCE_TIMEZONE: America/Sao_Paulo
    # Ao subir, declara ao Backend as fontes, a versão e os tipos de peça próprios, e só pega
    # trabalho depois de aceito: por isso espera o Backend saudável.
    depends_on:
      backend: { condition: service_healthy }

Valem duas lembranças. Os caminhos de um arquivo que se soma ao compose do produto partem da pasta do primeiro arquivo, não da do segundo. E os valores de RATE_LIMIT_PER_SECOND e CONCURRENCY de partida devem ser conservadores: meça antes de subir.

Para as questões de produção que o exemplo não cobre (ambiente de desenvolvimento com dados reais, publicação, evidência no ambiente), o adaptador real do repositório, adapters/mprs/, é a referência, e o processo de entrada de uma instituição está em apps/docs/content/integrar/playbook-novo-cliente.md.

Testes

Ao fim você sabe que testes escrever, e em que camada cada um pega o defeito certo.

Camada	O que prova	Como	Arquivo do exemplo
Mapeador	A resposta gravada vira o envelope esperado, e o corpo ilegível é recusado	Função pura, com corpos gravados em arquivo	test_example_mappers.py
Cliente	Rota, credencial, e o desfecho HTTP de cada falha no tipo certo	httpx.MockTransport	test_example_clients.py
Conformidade	O adaptador inteiro fala o contrato, de ponta a ponta	run_adapter contra o Backend falso, com a fonte simulada	test_example_conformance.py
Forma	Só o kit, as bibliotecas que ele instala e a biblioteca padrão; cliente sem tradução; mapeador sem E/S, e sem carregar o vocabulário (que lê arquivo); vocabulário só com chaves aceitas; imagem sem o token	Guardas sobre importações e arquivos	test_example_shape.py

Mapeador: grave o corpo que a fonte devolve, num arquivo, e afirme o que o contrato e as regras do produto pedem do envelope, não o que o código faz. O caso abaixo afirma que a palavra da fonte vira conceito sem perder a palavra original:

examples/example_adapter/tests/test_example_mappers.py

def test_a_source_word_becomes_a_national_concept_and_keeps_the_original_word(
    vocabulary: Vocabulary, payload: Callable[[str], bytes]
) -> None:
    proceeding = _proceeding(_ledger(vocabulary, payload("ledger_proceeding.json")))

    assert proceeding["confidentiality"] == {
        "vocabulary": "confidentiality",
        "label": "RESERVADO",
        "concept": "segredo_de_justica",
    }
    assert proceeding["confidentiality_rank"] == 1
    assert proceeding["procedural_class"] == {
        "vocabulary": "procedural_class",
        "label": "CIVEL COMUM",
        "concept": "7",
    }

Cliente: sirva a fonte por um transporte simulado e confira o que sai e o que volta. O teste do login confere que ele acontece uma vez só, e que o token vai no cabeçalho:

examples/example_adapter/tests/test_example_clients.py

async def test_the_archive_logs_in_once_and_sends_the_bearer_token(
    settings: ExampleSettings,
) -> None:
    seen: list[httpx.Request] = []

    def handler(request: httpx.Request) -> httpx.Response:
        if request.url.host == "login.archive.example":
            return httpx.Response(200, content=TOKEN)
        return httpx.Response(200, content=b"{}")

    client = _archive(settings, handler, seen)

    await client.fetch_proceeding("archive", "ARQ-1")
    await client.fetch_proceeding("archive", "ARQ-2")

    logins = [request for request in seen if request.url.host == "login.archive.example"]
    fetches = [request for request in seen if request.url.host == "archive.example"]
    assert len(logins) == 1
    assert b"client_secret=segredo" in logins[0].content
    assert [request.headers["authorization"] for request in fetches] == [
        "Bearer token-do-arquivo"
    ] * 2

Conformidade: é o teste do começo rápido, e é o que vale antes de qualquer instalação: você usa o run_adapter, que traz o Backend falso, as filas e o laço do kit. O que ele devolve é o que chegou ao Backend:

src/proceedings_adapter_kit/testing/fake_backend.py

    declarations: list[Json] = field(default_factory=list)
    ingested: list[Json] = field(default_factory=list)
    failures: list[Failure] = field(default_factory=list)
    uploads: dict[str, bytes] = field(default_factory=dict)
    violations: list[Violation] = field(default_factory=list)
    reported: set[int] = field(default_factory=set)
    calls: list[tuple[str, str]] = field(default_factory=list)

Para um caso de falha, afirme o relato por trabalho. O exemplo mostra um de cada tipo:

examples/example_adapter/tests/test_example_conformance.py

    result.assert_conforming()
    assert {e["job_id"]: e["source_system"] for e in result.ingested} == {1: "ledger", 2: "archive"}
    assert {e["adapter_version"] for e in result.ingested} == {ADAPTER_VERSION}
    assert {failure.job_id: failure.kind for failure in result.failures} == {
        4: "not_found",
        5: "transient",
        6: "invalid",
        7: "invalid",
        8: "not_found",
    }
    assert "recusou: HTTP 400" in next(f.error for f in result.failures if f.job_id == 6)
    assert "fora do formato do Livro" in next(f.error for f in result.failures if f.job_id == 7)

Para rodar tudo, da pasta do kit:

uv sync --locked && uv run ruff check && uv run ruff format --check && uv run mypy . && uv run pytest -q

Veja o teste falhar

Quebre de propósito o que o seu teste promete pegar: troque uma condição, tire um efeito, e veja o teste falhar. Teste que nunca falhou não prova nada.

Referência dos protocolos

Ao fim você tem à mão a forma exata de cada peça que o kit espera de você, e da suíte que o confere.

O que você entrega ao kit

O cliente da fonte só faz E/S. Cada busca levanta uma das três falhas da fonte, e flow é a vazão que o kit relata:

src/proceedings_adapter_kit/source/protocols.py

class SourceClient(Protocol):
    """Só E/S. Cada busca levanta `SourceNotFoundError`, `TransientSourceError` ou
    `SourceRejectedError`, e `flow` é a vazão que o kit relata."""

    @property
    def flow(self) -> Flow: ...

    async def fetch_proceeding(self, source_key: str, number: str) -> Payload: ...

    async def fetch_filing(
        self, source_key: str, number: str, filing_source_identifier: str
    ) -> FilingFile: ...

O mapeador é uma função pura. O kit lhe passa o contexto do trabalho, e o vocabulário e a versão do adaptador se prendem à função com functools.partial:

src/proceedings_adapter_kit/source/protocols.py

@dataclass(frozen=True, slots=True)
class MapContext:
    """O que o kit sabe do job e passa à tradução: a chave da fonte, o job que o envelope leva e o
    fuso em que a fonte grava as datas."""

    source_key: str
    job_id: int
    timezone: ZoneInfo


# Pura, sem E/S: o payload da fonte vira o envelope, ou levanta `InvalidPayloadError`. As tabelas
# de tradução e a versão do adaptador são da instituição: ela as prende na função, com `partial`.
type Mapper = Callable[[Payload, MapContext], Json]

Uma fonte junta o cliente e o mapeador, e o adaptador junta a declaração e uma fonte por chave:

src/proceedings_adapter_kit/source/protocols.py

@dataclass(frozen=True, slots=True)
class Source:
    client: SourceClient
    mapper: Mapper


@dataclass(frozen=True, slots=True)
class Adapter:
    """O que o adaptador entrega ao kit: a declaração e uma fonte por chave declarada."""

    declaration: Declaration
    sources: Mapping[str, Source]

As falhas da fonte

src/proceedings_adapter_kit/source/errors.py

class SourceError(Exception):
    """Falha ao buscar na fonte."""


class SourceNotFoundError(SourceError):
    """A fonte disse que o processo não existe. Repetir não muda a resposta."""


class TransientSourceError(SourceError):
    """Limite, falha do servidor, tempo esgotado ou credencial indisponível: vale repetir."""


class SourceRejectedError(SourceError):
    """A fonte recusou de um jeito que repetir não conserta, ou respondeu fora do formato."""


class InvalidPayloadError(Exception):
    """A tradução não conseguiu ler o que a fonte devolveu. Repetir não muda a resposta: o job
    para já, relatado como `invalid`."""

O pedido à fonte

src/proceedings_adapter_kit/source/http.py

    def __init__(
        self,
        http: httpx.AsyncClient,
        *,
        rate_limit_per_second: float,
        timeout_seconds: float,
        label: str,
    ) -> None:
        self._http = http
        self._timeout = timeout_seconds
        self._label = label
        # Um pedido a cada 1/taxa segundos, sem rajada; a taxa se adapta até a quota.
        self.flow = Flow(rate_limit_per_second)
        # Limite da fonte pausa todos os pedidos deste processo. A espera da pausa acorda na parada.
        self._throttle = Throttle(self.flow.sleep)

    async def get(
        self,
        url: str,
        *,
        headers: Mapping[str, str] | None = None,
        not_found: int = 404,
        not_found_message: str,
    ) -> httpx.Response:
        """`not_found` é o status com que a fonte diz que o recurso não existe."""

A configuração comum

src/proceedings_adapter_kit/settings.py

    backend_url: AnyHttpUrl
    # O mesmo segredo que o Backend lê em INGESTION_SERVICE_TOKEN
    ingestion_service_token: SecretStr

    # Sem valor padrão: a fonte não publica limite e nada foi medido. A instalação decide.
    rate_limit_per_second: Annotated[float, Field(gt=0)]
    # Fuso em que a fonte grava as datas sem offset. Declarado uma vez, aqui.
    source_timezone: str

    concurrency: Annotated[int, Field(ge=1, le=64)] = 4
    batch_size: Annotated[int, Field(ge=1, le=100)] = 10
    idle_seconds: Annotated[float, Field(gt=0)] = 2.0
    # Por leitura, não pelo pedido inteiro: uma fonte medida levou mais de 30 s para mandar o
    # primeiro byte de um documento de 250 MB ou mais. A conexão cai em 5 s, à parte.
    request_timeout_seconds: Annotated[float, Field(gt=0)] = 300.0
    worker: str = Field(default_factory=_worker_name, min_length=1, max_length=200)

Subir o adaptador

src/proceedings_adapter_kit/run.py

async def run(
    adapter: Adapter,
    settings: KitSettings,
    *,
    backend_transport: httpx.AsyncBaseTransport | None = None,
    upload_transport: httpx.AsyncBaseTransport | None = None,
) -> None:

A suíte de conformidade

src/proceedings_adapter_kit/testing/plugin.py

class RunAdapter(Protocol):
    async def __call__(
        self,
        adapter: Adapter,
        *,
        jobs: Sequence[Mapping[str, object]],
        settings: KitSettings | None = None,
        timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS,
    ) -> FakeBackend: ...

Atributo do FakeBackend	O que guarda
declarations	Cada declaração aceita
ingested	Cada envelope aceito
failures	Cada relato de falha (job_id, kind, error)
uploads	Os bytes de cada peça, pelo filing_id
violations	Cada mensagem ou envio recusado, com a regra
reported	Os trabalhos que o adaptador terminou, por qualquer via
calls	Cada chamada, na ordem, como método e rota

assert_conforming() falha listando as violações:

src/proceedings_adapter_kit/testing/fake_backend.py

    def assert_conforming(self) -> None:
        """Falha, listando regra e motivo, se alguma mensagem ou envio foi recusado."""
        if self.violations:
            listed = "; ".join(f"{v.rule}: {v.detail}" for v in self.violations)
            raise AssertionError(f"o adaptador violou o contrato: {listed}")

As seis regras que o Backend aplica ao envelope além do esquema têm um corpus de casos, em src/proceedings_adapter_kit/testing/cases/, que vai no pacote do kit. Cada arquivo é um envelope inteiro e diz a regra que quebra, ou que é aceito. O kit julga todos pelo Backend falso, e o portão da plataforma os julga pelo modelo do Backend de verdade, e falha, nomeando o caso, se os dois discordarem. load_cases(), em proceedings_adapter_kit.testing.corpus, os devolve, se você quiser rodá-los contra o seu mapeador.

Regra	O que recusa	Casos do corpus
term-code-or-label	Termo sem code e sem label, com a string vazia contando como ausente	term-code-or-label, term-code-or-label-nested
party-name-or-document	Parte sem name e sem document_number	party-name-or-document
timezone	filed_at, docketed_at, occurred_at ou created_at sem fuso, ou escrita que o Backend não lê: deslocamento curto (+03), formato básico, data de semana ISO e 24:00	timezone-filed-at, timezone-docketed-at, timezone-occurred-at, timezone-created-at, timezone-short-offset, timezone-basic-format, timezone-week-date, timezone-24-hours
reference	represents_ref que não é uma parte, ou event_refs que não é um evento	reference-represents-ref, reference-event-refs
capture-base64	content da captura que não é base64	capture-base64, capture-base64-padding, capture-base64-stray-character, capture-base64-not-ascii
capture-digest	digest que não confere com o conteúdo, ou com algoritmo que não é sha256 nem sha512	capture-digest, capture-digest-algorithm

O contrato

Arquivo	Versão	O que é
ingestion.json	2	O envelope, o pedido de lote, o trabalho tomado e o relato de falha
source-systems.json	4	A declaração: fontes, versão do adaptador e tipos de peça próprios
filing-upload.json	2	O pedido de ticket e a conclusão do envio da peça
concept-keys.json		As chaves de conceito que o envelope pode citar, por vocabulário

As cópias fixadas ficam em src/proceedings_adapter_kit/contract/schemas/, e o kit confere que as suas mensagens têm a forma delas. Contrato novo pede kit novo.

Quando o guia não basta

Se você precisou de algo que o guia não explica, o guia tem um defeito: corrija-o junto com o adaptador, no mesmo trabalho. Cada dúvida de quem usa o kit pela primeira vez é uma seção que falta.