5  Guia do desenvolvedor do colibri

Este guia detalha o ritual de desenvolvimento de pipelines de ingestão e modelos dbt no repositório Colibri, assumindo que você já seguiu o Guia de instalação.

5.1 Camadas do dbt

Os modelos dbt vivem em dbt/modelos/ (note: modelos, não models), organizados em três camadas, cada uma com um subdiretório por fonte de dados (ex.: pncp_comprasgov/, ncm/, catmats/, nfe_cgu/, margem_preferencia/, tradutor_catmat_ncm/):

Camada Materialização padrão Schema Convenção de nome
staging/ table staging stg_<fonte>__<entidade>.sql
intermediate/ view intermediate int_<fonte>__<entidade>.sql
marts/ view marts mrt_<fonte>__<entidade>.sql

Esses padrões são definidos globalmente em dbt/dbt_project.yml e podem ser sobrescritos por modelo, via bloco { config(...) } no início do arquivo SQL — por exemplo, os modelos de staging do comprasGOV usam materialized='incremental' em vez do table padrão da camada.

5.2 Contratos e testes

Use contract={'enforced': true} no config() de um modelo quando quiser que o dbt valide o schema (tipos e ordem de colunas) em tempo de build — hoje isso só é usado nos modelos de tradutor_catmat_ncm, como referência.

Testes de dados usam a sintaxe data_tests (não a antiga tests) nos arquivos _<fonte>__models.yml:

- name: catmat
  data_type: integer
  description: Código CATMAT do item de material.
  data_tests:
    - not_null
    - unique

O teste genérico unique (e outros testes de coluna) aceita uma expressão em column_name, não só o nome literal de uma coluna — útil para testar unicidade de chaves compostas:

data_tests:
  - unique:
      column_name: ncm || '-' || tipo_margem

Ao usar argumentos em testes genéricos (como accepted_values), lembre-se de que este projeto roda em dbt-fusion, que exige os argumentos aninhados sob arguments::

data_tests:
  - accepted_values:
      arguments:
        values: ['humano', 'ia']

Hoje, data_tests só está presente nos modelos de tradutor_catmat_ncm — ao adicionar ou revisar um modelo em outra fonte, é uma boa oportunidade para acrescentar testes de not_null/unique nas colunas-chave.

5.3 O padrão SCD2 (histórico versionado)

Os modelos incrementais do comprasGOV (compras, itens, resultados) seguem um padrão slowly changing dimension tipo 2:

  1. Deduplicar registros do mesmo período usando ROW_NUMBER() particionado pela chave natural, preferindo a linha mais completa (via uma macro tipo contar_colunas_preenchidas).
  2. Calcular valido_de/valido_ate/is_current com LEAD() sobre a janela ordenada pela chave, marcando quando cada versão de um registro deixou de valer.

Esse padrão é o que permite consultar o estado de uma compra “como ela estava” em uma data específica, e não só o estado mais recente.

5.4 Ritual de scaffolding com codegen

O pacote dbt-labs/codegen (listado em dbt/packages.yml) é usado para gerar o esqueleto de documentação de colunas de um modelo novo:

cd dbt
dbt run-operation generate_model_yaml --args '{"model_names": ["stg_nova_fonte__entidade"]}'
cd ..

Cole o YAML gerado no arquivo _<fonte>__models.yml correspondente e complete as descrições e testes manualmente.

5.5 Adicionar uma pipeline de ingestão nova

Cada fonte de dados tem um pacote próprio em ingestion/<fonte>/, com dois arquivos:

  • extract.py: baixa/parseia os dados brutos da fonte. Usa um manifesto CSV (ex.: ncm_manifesto.csv) com o hash SHA-256 do conteúdo extraído (excluindo campos de data) para decidir se há algo novo a extrair — se o hash não mudou, a extração é pulada. O manifesto é sincronizado com o bucket para persistir entre execuções e ambientes.
  • pipeline.py: orquestra a extração e o dbt. Expõe uma função main(bucket=None) que carrega o segredo de desenvolvedor, baixa o catálogo, roda executar_ingestao(), e — só se houve mudança — dispara dbt run --select <modelos da fonte> via subprocess, e por fim reenvia o manifesto e o catálogo atualizados para o bucket. Essa ordem (dbt roda antes do reupload) garante que o estado do bucket só avança se o dbt run tiver sucesso.

Para que uma fonte nova apareça em colibri pipeline run --apenas <fonte>, registre-a em três pontos de cli.py (raiz do repo):

  1. Nas opções (choices) do parâmetro --apenas do comando pipeline run.
  2. No dicionário pipelines = {...} que mapeia o nome da fonte ao módulo ingestion.<fonte>.pipeline.
  3. Em _MANIFESTOS, se a fonte gerar um manifesto que precise ser sincronizado por colibri sincronizar.

Helpers compartilhados entre pipelines vivem em utils/ (raiz do repo, não ingestion/utils/): carregar_segredo.py, criar_cliente.py, baixar_catalogo.py, manifesto_bucket.py, salvar_arquivo_no_bucket.py, csv_para_parquet.py, ducklake.py, entre outros.

5.6 Checklist antes de abrir um PR

O template de PR exige rodar, na raiz do repositório:

ruff format .
yamlfix .
pytest
cd dbt && ../env/bin/dbt test --project-dir . --profiles-dir . --target prod && cd ..

Note que o dbt test do checklist roda contra o target prod, usando o binário do dbt instalado no próprio ambiente virtual do projeto (env/bin/dbt) — não o dbt do PATH do sistema.