dlt na Prática

Construindo pipelines de ingestão em Python do zero ao produção

Um Artigo técnico sobre a biblioteca que virou o padrão de fato para mover dados em Python — com um laboratório executável do começo ao fim.

Por que este artigo existe

Quem trabalha com dados chega, mais cedo ou mais tarde, no mesmo problema chato: você tem dados espalhados em lugares diferentes — uma API REST aqui, um bucket S3 ali, um Postgres legado acolá, uma planilha que alguém insiste em manter — e precisa juntar tudo num lugar onde dê para analisar. Essa é a parte “EL” do ELT (Extract e Load), e historicamente ela sempre foi resolvida com scripts Python frágeis, cheios de requests, json.loads, tratamento de paginação na unha e aquele CREATE TABLE IF NOT EXISTS copiado de outro projeto.

O dlt (data load tool) existe para matar essa categoria inteira de código boilerplate. Ele é uma biblioteca Python leve, open-source, que cuida de extrair dados de fontes bagunçadas, normalizar o schema automaticamente e carregar num destino estruturado — sem que você precise escrever DDL, gerenciar tipos ou reinventar paginação a cada projeto.

E não é hype de nicho: para dar dimensão, a comunidade dlt saiu de cerca de 2.400 pipelines criados em janeiro de 2025 para 81.000 em janeiro de 2026 — um crescimento de 34x em um ano. Boa parte disso vem de agentes de IA gerando pipelines automaticamente, justamente porque a “forma” do dlt é declarativa e previsível o suficiente para uma LLM acertar de primeira. Voltamos a esse ponto no final.

Uma distinção que evita confusão logo de cara: dlt não é dbt. São ferramentas com nomes parecidos e propósitos opostos. O dlt cuida do EL — trazer o dado bruto de fora para dentro. O dbt (e o SQLMesh) cuidam do T — transformar o dado que já está dentro do warehouse em modelos de negócio. Elas são vizinhas de pipeline, não concorrentes. Ao longo do artigo isso fica claro; guarde a régua por enquanto.

Este artigo é a linha “do zero ao expert”: começamos com o conceito, montamos um laboratório que você roda no terminal, e vamos subindo até incremental loading, merge, transformações, segredos, troca de destino e deploy.

Nota de versão: o lab foi escrito para dlt ≥ 1.21 e validado com dlt 1.30.0 + DuckDB + pandas. A API de fonte REST (rest_api_source) está estável nessa linha. Se você veio das versões 0.x, algumas assinaturas mudaram — o código deste lab já usa a API atual.

O modelo mental do dlt

Antes de sair digitando, vale fixar quatro conceitos. Se você entender esses quatro, entendeu 90% do dlt.

Esses são os quatro conceitos centrais usados no artigo para formar o modelo mental do dlt.

O ciclo de vida de qualquer execução do dlt tem três fases, e quando você chama pipeline.run() as três acontecem em sequência:

1. Extract — puxa os dados brutos da fonte e guarda em disco (arquivos intermediários).
2. Normalize — infere o schema, achata estruturas aninhadas, resolve tipos, cria tabelas-filhas para listas.
3. Load — sobe os dados normalizados para o destino, respeitando a estratégia de escrita que você configurou.

Visualmente, o fluxo é este:

FONTE                    dlt pipeline.run()                      DESTINO
┌──────────┐      ┌──────────┬───────────┬────────┐         ┌──────────────┐
│ API REST │      │          │           │        │         │   DuckDB     │
│ SQL DB   │─────▶│ EXTRACT  │ NORMALIZE │  LOAD  │────────▶│   Postgres   │
│ Arquivos │      │ (→disco) │ (schema)  │ (→dest)│         │   S3 / MinIO │
│ Python   │      │          │           │        │         │   Snowflake  │
└──────────┘      └──────────┴───────────┴────────┘         └──────────────┘
                       resiliente e retomável        estado salvo no destino

O grande truque do dlt é que essas três fases são desacopladas e resilientes. Se o load falhar no meio, o estado do extract continua em disco e você pode retomar. Isso é muito diferente de um script requests + INSERT que morre no registro 4.999 de 10.000 e te deixa com meia tabela.

E há um detalhe que muita gente descobre tarde: o dlt guarda o próprio estado no destino, em tabelas de sistema com prefixo _dlt_ (_dlt_loads, _dlt_pipeline_state, _dlt_version). É isso que permite carga incremental idempotente sem você manter nenhum arquivo de controle na mão — o pipeline "lembra" onde parou consultando o próprio banco de destino. Vamos usar essa propriedade no Lab 4.

Estrutura do Projeto

lab-dlt/
├── .dlt/
│   ├── config.toml              # configs não sensíveis
│   └── secrets.toml.example     # modelo de segredos (copie → secrets.toml)
├── .gitignore
├── requirements.txt
├── README.md                    # atalho de setup
├── lab01_basico.py
├── lab02_rest.py
├── lab03_merge.py
├── lab04_incremental.py
├── lab05_transform.py
├── lab06_secrets.py
├── lab07_postgres.py
└── lab08_airflow_sketch.py

Cada lab gera o próprio arquivo *.duckdb (ex.: lab01.duckdb). Eles estão no .gitignore.

Uma sugestão de leitura: os labs são incrementais em dificuldade, mas independentes em execução. Você pode rodar o Lab 5 sem ter rodado o Lab 4. Se o seu tempo for curto, os labs 3 (merge) e 4 (incremental) são os que carregam o maior retorno sobre o investimento — é onde o dlt deixa de ser “conveniência” e vira “por que eu fazia isso na mão antes?”.

Laboratório 0 — Preparando o ambiente

O lab é autocontido. Use uv (recomendado) ou venv. Não instale dlt no Python do sistema.

Passo a passo

# Se você já está dentro do clone do repo:
cd lab-dlt

# Cria o ambiente virtual
uv venv
source .venv/bin/activate        # Linux/Mac
# .venv\Scripts\activate         # Windows
# Instala dlt com DuckDB + pandas (necessário para .df())
uv pip install -r requirements.txt
# Equivalente manual:
# uv pip install "dlt[duckdb]" pandas

O requirements.txt do projeto fixa:

dlt[duckdb]>=1.21.0
pandas>=2.0.0

Confirme a instalação:

dlt --version
# esperado (ambiente de validação): dlt 1.30.0

Estrutura de configuração

Já vem no repo:

ls .dlt/
# config.toml
# secrets.toml.example

Conteúdo de .dlt/config.toml:

# Configurações não sensíveis do dlt
# Segredos ficam em secrets.toml (nunca comitar)

[runtime]
log_level = "WARNING"

Para os Labs 6 e 7 (segredos / Postgres), prepare o arquivo local:

cp .dlt/secrets.toml.example .dlt/secrets.toml
# edite api_token e, se for usar Postgres, as credenciais

O .gitignore já ignora .dlt/secrets.toml, *.duckdb e a pasta do venv.

Critério de sucesso do Lab 0

• [ ] source .venv/bin/activate funciona
• [ ] dlt --version imprime uma versão ≥ 1.21
• [ ] python -c "import dlt, pandas; print('ok')" não falha

Laboratório 1 — O “Hello World”: uma lista Python vira tabela

Antes de complicar com APIs, vemos o dlt no formato mais nu. O objetivo aqui é sentir na mão a promessa central da ferramenta — dados entram, tabela sai, você não escreveu DDL — para reconhecê-la depois quando a fonte for complicada.

Código (lab01_basico.py)

"""Laboratório 1 — Hello World: uma lista Python vira tabela."""

import dlt
# Nossos "dados de origem" - uma lista de dicionários Python.
dados = [
    {"id": 1, "nome": "Ana", "cidade": "Recife"},
    {"id": 2, "nome": "Bruno", "cidade": "Belo Horizonte"},
    {"id": 3, "nome": "Carla", "cidade": "Manaus"},
]
# 1. Cria o pipeline: dá nome, escolhe destino e o dataset (schema)
pipeline = dlt.pipeline(
    pipeline_name="lab01",
    destination="duckdb",
    dataset_name="cadastro",
)
# 2. Roda: extrai, normaliza e carrega na tabela "pessoas"
load_info = pipeline.run(dados, table_name="pessoas")
print(load_info)
print(pipeline.dataset().pessoas.df())

Execução

python lab01_basico.py

O que observar

O dlt cria lab01.duckdb na pasta do projeto. Você não escreveu CREATE TABLE: ele inferiu tipos (id inteiro, nome/cidade texto), criou a tabela pessoas no schema cadastro e inseriu os três registros.

Além das suas colunas, aparecem metadados _dlt_load_id (qual execução carregou a linha) e _dlt_id (identificador único de linha). Esses dois campos não são enfeite: o _dlt_load_id é o que torna auditável responder "quando essa linha entrou e por qual execução?", e o _dlt_id é a âncora que liga tabelas-pai e tabelas-filha quando há dados aninhados (você vê isso já no Lab 2).

Resultado esperado (resumo)

Critério de sucesso

• [ ] Arquivo lab01.duckdb existe
• [ ] DataFrame com 3 linhas e colunas id, nome, cidade, _dlt_*

Laboratório 2 — Consumindo uma API REST de verdade

Cenário realista: extrair de uma API REST com rest_api_source. Usamos a JSONPlaceholder — API pública de teste, sem credencial.

Código (lab02_rest.py)

"""Laboratório 2 — Consumindo uma API REST (JSONPlaceholder)."""
import dlt
from dlt.sources.rest_api import rest_api_source
# Definição declarativa da fonte REST
source = rest_api_source(
    {
        "client": {
            "base_url": "https://jsonplaceholder.typicode.com/",
        },
        "resources": [
            "posts",
            "users",
        ],
    }
)
pipeline = dlt.pipeline(
    pipeline_name="lab02_jsonplaceholder",
    destination="duckdb",
    dataset_name="blog",
)
load_info = pipeline.run(source)
print(load_info)
# Lista as tabelas do dataset e mostra os dados
dataset = pipeline.dataset()
print(dataset.users.df())
print(dataset.posts.df().head())

Execução

python lab02_rest.py

Requer rede. Pode aparecer um warning de paginator (SinglePagePaginator) — normal: a JSONPlaceholder devolve a lista inteira numa página só.

O que o dlt fez

• Requisições HTTP para /posts e /users
• Inferência de schema
• Criação das tabelas no DuckDB (lab02_jsonplaceholder.duckdb)

Repare no tamanho do que você escreveu: uma URL base e dois nomes de recurso. Não há requests, não há laço de paginação, não há tratamento de tipos. Essa concisão é proposital — e é a mesma "forma" que uma LLM consegue preencher com alta taxa de acerto, ponto que retomamos no fim.

Pontos de atenção no output de users

1. Achatamento (flattening). Objetos aninhados viram colunas com __: address__street, address__geo__lat, address__geo__lng.
2. Listas aninhadas viram tabela-filha, ligada por _dlt_parent_id (cardinalidade 1:N).
3. Sem PK automática. A fonte REST não define chave primária por padrão — isso entra no Lab 3.

A diferença entre os itens 1 e 2 é sutil e vale internalizar: um objeto aninhado (um address único por usuário) achata em colunas na mesma linha; uma lista aninhada (vários itens por registro-pai) vira uma tabela separada, porque manter N itens numa célula quebraria a modelagem relacional. O dlt decide isso sozinho, preservando a cardinalidade — é um dos comportamentos mais elegantes da ferramenta e algo que, na mão, dá muito retrabalho.

Critério de sucesso

• [ ] load_info sem jobs falhos
• [ ] users com 10 linhas; posts com 100 linhas (API padrão)

Intervalo conceitual — schema evolution, o superpoder silencioso

Antes de seguir para o merge, vale parar num comportamento que o dlt executa tão discretamente que é fácil não perceber que ele existe — até o dia em que ele te salva.

APIs e bancos de origem mudam. Um dia o /users passa a devolver um campo novo, discount. Num pipeline caseiro, isso costuma dar um de dois desfechos ruins: ou o script quebra, ou o campo novo é silenciosamente ignorado e some. No dlt, o padrão é um terceiro caminho: schema evolution. Na próxima execução, ele detecta a coluna nova, adiciona ao schema do destino e segue carregando, sem quebrar e sem perder o dado.

E quando o conflito é de tipo — um campo que era numérico passa a vir como texto — o dlt também não derruba a carga nem sobrescreve o que existe. Ele cria uma coluna variante, seguindo o padrão <coluna>__v_<tipo> (por exemplo, amount__v_text), preservando a coluna original intacta. Nenhuma linha é descartada.

Isso resolve o problema de robustez, mas cria um de governança: as mudanças acontecem em silêncio. Como saber que o schema mexeu? A trilha de auditoria são as próprias tabelas _dlt_: a _dlt_version registra cada versão do schema, e a _dlt_loads amarra cada carga à versão vigente. Dá para monitorar isso e alertar os donos da tabela quando algo muda.

E se, em vez de aceitar tudo, você quiser controlar a evolução — porque aquela tabela alimenta um relatório regulatório e ninguém pode adicionar coluna sem revisão? Aí entram os schema contracts. Você define, por entidade (tables, columns, data_type), um modo:

Para um pipeline de dados de saúde ou financeiro — onde uma coluna fantasma em produção é um problema de conformidade, não só de qualidade — congelar o schema com freeze faz a ingestão falhar alto e cedo, deixando a equipe decidir conscientemente se corrige a origem ou evolui o contrato. É o oposto de descobrir a coluna misteriosa três semanas depois. Não vamos codar contrato nos labs (para não desviar do fio principal), mas saber que a alavanca existe é o que separa usar o dlt de operá-lo em produção.

Laboratório 3 — Chaves primárias, merge e a estratégia de escrita

Rode o Lab 2 duas vezes e a contagem de posts dobraria: a estratégia padrão é append. Em produção isso quase nunca é o que você quer.

A escolha entre as três não é detalhe de configuração — é uma decisão de modelagem. Ela responde à pergunta “o que significa rodar este pipeline de novo?”. Para uma tabela de eventos que só cresce, append é correto. Para uma dimensão de clientes que muda, merge é o único que mantém a tabela fiel à origem sem inchar. Errar aqui é a causa nº 1 de tabela duplicada em pipeline de iniciante.

Código (lab03_merge.py)

"""Laboratório 3 — Chaves primárias, merge e write disposition."""
import dlt
from dlt.sources.rest_api import rest_api_source
source = rest_api_source(
    {
        "client": {
            "base_url": "https://jsonplaceholder.typicode.com/",
        },
        # resource_defaults aplica configurações a TODOS os recursos de uma vez
        "resource_defaults": {
            "write_disposition": "merge",
            "primary_key": "id",
        },
        "resources": [
            "posts",
            "users",
        ],
    }
)
pipeline = dlt.pipeline(
    pipeline_name="lab03_merge",
    destination="duckdb",
    dataset_name="blog",
)
load_info = pipeline.run(source)
print(load_info)
print("Posts:", len(pipeline.dataset().posts.df()))

Execução e teste de idempotência

python lab03_merge.py
python lab03_merge.py

Resultado esperado

Nas duas execuções:

O merge fez upsert: id já existente atualiza no lugar, não duplica. Isso é idempotência — rode mil vezes, o resultado é o mesmo. É a propriedade que deixa você reexecutar um pipeline com tranquilidade depois de uma falha, sem medo de duplicar dado. Sem ela, "rodar de novo" é uma aposta.

O resource_defaults aplica write_disposition e primary_key a todos os recursos; um recurso específico pode sobrescrever depois.

Critério de sucesso

• [ ] Primeira e segunda execução imprimem Posts: 100
• [ ] Arquivo lab03_merge.duckdb criado

Laboratório 4 — Incremental loading: só o que mudou

Feature que, sozinha, justifica adotar o dlt: carga incremental. Em vez de baixar a base inteira, só o que é novo ou alterado desde a última execução. O dlt gerencia o estado do cursor.

Usamos a API pública do GitHub (eventos do repositório duckdb/duckdb). O cursor é created_at.

Código (lab04_incremental.py)

"""Laboratório 4 — Incremental loading com a API do GitHub."""
import dlt
from dlt.sources.rest_api import rest_api_source
source = rest_api_source(
    {
        "client": {
            "base_url": "https://api.github.com/",
            # A API do GitHub usa o cabeçalho Link para paginar
            "paginator": "header_link",
        },
        "resource_defaults": {
            "write_disposition": "merge",
            "primary_key": "id",
        },
        "resources": [
            {
                "name": "eventos_repo",
                "endpoint": {
                    "path": "repos/duckdb/duckdb/events",
                    "params": {
                        "per_page": 100,
                        # Cursor incremental: só traz registros mais recentes
                        # que o último created_at visto.
                        "since": {
                            "type": "incremental",
                            "cursor_path": "created_at",
                            "initial_value": "2024-01-01T00:00:00Z",
                        },
                    },
                },
            },
        ],
    }
)
pipeline = dlt.pipeline(
    pipeline_name="lab04_github",
    destination="duckdb",
    dataset_name="github",
)
load_info = pipeline.run(source)
print(load_info)
print("Eventos carregados:", len(pipeline.dataset().eventos_repo.df()))

Execução

python lab04_incremental.py
python lab04_incremental.py   # imediatamente depois

O que esperar

• 1ª execução: baixa eventos a partir de initial_value. Na validação deste repo: 298 eventos (o número muda com o tempo — o GitHub só retém os eventos recentes).
• 2ª execução: poucos ou zero registros novos. Mensagem típica: Pipeline lab04_github load step did not start / 0 load package(s) — o cursor já está no máximo visto. A contagem total permanece a mesma.

Warnings sobre colunas sem tipo inferido (payload__... vazios) são esperados no schema do GitHub Events; não impedem a carga.

Get Hugo Habbema’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

Pare um segundo no que acabou de acontecer entre as duas execuções: você não escreveu nenhuma linha para guardar “qual foi o último created_at". O dlt gravou esse marcador na tabela _dlt_pipeline_state, no próprio destino, e o releu sozinho na segunda rodada. Todo pipeline caseiro acaba desenvolvendo uma versão precária disso — um arquivo .txt com a última data, uma tabela de controle à parte, uma variável no Airflow. O dlt te entrega esse gerenciamento de estado de graça, e à prova de execução em ambiente efêmero.

Boa prática: combine sempre incremental + write_disposition: "merge" + primary_key correta. Com append, um registro de borda repetido (cursor >=) duplica.

O estado fica no destino (tabela _dlt_pipeline_state). Você não precisa de arquivo "última data processada" na mão.

Um nome para o que você montou

Isso tem nome em engenharia de dados: é uma carga incremental baseada em cursor com merge. É o padrão dominante para sincronizar APIs e bancos transacionais para um lakehouse ou warehouse. Vale guardar o termo — é o coração de uma fatia enorme dos pipelines de ingestão reais, e o dlt te dá isso em cinco linhas de configuração declarativa.

Critério de sucesso

• [ ] 1ª run carrega N > 0 eventos
• [ ] 2ª run imediata não aumenta N (ou aumenta só o que chegou de fato no intervalo)

Laboratório 5 — Filtrando e transformando na ingestão

Transformações leves antes do load: filtrar, normalizar, anonimizar. Mecanismo: processing_steps.

• filter — retorna True/False; False descarta o registro.
• map — recebe o registro e devolve o registro transformado.

Código (lab05_transform.py)

"""Laboratório 5 — Filtrando e transformando na ingestão."""
import dlt
from dlt.sources.rest_api import rest_api_source

def normalizar_email(registro):
    """Passo de MAP: força e-mail para minúsculas."""
    if "email" in registro and registro["email"]:
        registro["email"] = registro["email"].lower()
    return registro

def anonimizar_telefone(registro):
    """Passo de MAP: mascara o telefone, guardando só os 4 últimos dígitos."""
    tel = registro.get("phone", "")
    digitos = "".join(c for c in tel if c.isdigit())
    if len(digitos) >= 4:
        registro["phone"] = "****" + digitos[-4:]
    return registro

source = rest_api_source(
    {
        "client": {
            "base_url": "https://jsonplaceholder.typicode.com/",
        },
        "resource_defaults": {
            "write_disposition": "merge",
            "primary_key": "id",
        },
        "resources": [
            {
                "name": "users",
                "endpoint": {"path": "users"},
                # As transformações rodam na ordem em que aparecem na lista
                "processing_steps": [
                    # 1. Só mantém usuários com id ÍMPAR
                    {"filter": lambda x: x["id"] % 2 != 0},
                    # 2. E-mail em minúsculas
                    {"map": normalizar_email},
                    # 3. Anonimiza o telefone
                    {"map": anonimizar_telefone},
                ],
            },
            # posts continua vindo cru, sem transformação
            "posts",
        ],
    }
)
pipeline = dlt.pipeline(
    pipeline_name="lab05_transform",
    destination="duckdb",
    dataset_name="blog_tratado",
)
load_info = pipeline.run(source)
print(load_info)
df = pipeline.dataset().users.df()
print(df[["id", "email", "phone"]])

Execução

python lab05_transform.py

Resultado esperado

id                     email     phone
0   1         sincere@april.biz  ****6442
1   3        nathan@yesenia.net  ****4447
2   5  lucio_hettinger@annie.ca  ****1289
3   7    telly.hoeger@billy.biz  ****6132
4   9   chaim_mcdermott@dana.io  ****1206

Só ids ímpares, e-mails em minúsculas, telefones mascarados. Na mesma lista de resources misturamos dicionário completo (users) e string simples (posts) — você só paga a complexidade do dicionário onde realmente precisa dela.

Arquitetura: processing_steps rodam em streaming, durante a extração. A distinção que importa é o que pertence aqui e o que não. Higiene estrutural e de conformidade — mascarar um dado sensível antes de ele tocar o disco, descartar registros inválidos, padronizar formatos — pertence à porta de entrada, e o exemplo de anonimização de telefone acima não é acadêmico: para quem trabalha com dado de saúde, anonimizar na ingestão é o que impede o dado bruto identificável de nem chegar a existir no destino. Já joins, agregações e modelagem dimensional não pertencem aqui — isso é trabalho da camada T (dbt, SQLMesh, SQL no warehouse), onde há o conjunto inteiro para operar, não um registro de cada vez.

Critério de sucesso

• [ ] Exatamente 5 usuários (ids 1, 3, 5, 7, 9)
• [ ] Telefones no formato ****NNNN

Laboratório 6 — Credenciais e segredos do jeito certo

Na vida real quase tudo pede autenticação. Hardcodar token no .py e commitar é o pior caminho — e o mais comum. O dlt tem config/segredos próprios justamente para tornar o caminho certo mais fácil que o errado.

Arquivo Conteúdo .dlt/config.toml Não sensível (URLs, log level) .dlt/secrets.toml Tokens, senhas — no .gitignore

Preparação

cp .dlt/secrets.toml.example .dlt/secrets.toml

Modelo (.dlt/secrets.toml.example):

# Copie para secrets.toml e preencha os valores.
# NUNCA comite secrets.toml no Git.

[sources.github]
# Token pessoal do GitHub (escopo público basta para o lab)
api_token = "ghp_seu_token_pessoal_aqui"
# Descomente e preencha para o Lab 07 com --postgres
# [destination.postgres.credentials]
# database = "analytics"
# username = "dlt_loader"
# password = "senha_secreta"
# host = "localhost"
# port = 5432

Em produção, prefira variável de ambiente (mesma hierarquia, achatada):

export SOURCES__GITHUB__API_TOKEN="ghp_seu_token"

Cascata do dlt: código → env var → secrets.toml → config.toml.

O detalhe que faz essa cascata valer ouro: o mesmo código roda em todos os cenários. No seu notebook, ele lê o secrets.toml. No container de produção, lê a variável de ambiente injetada pelo orquestrador ou pelo secret manager — sem uma linha de diferença no .py. Você nunca escolhe "modo dev" ou "modo prod" no código; o dlt procura o segredo em cascata e usa o primeiro que encontra.

Código (lab06_secrets.py)

No projeto, o lab não quebra sem token: cai no rate limit público, com aviso. Com token, usa bearer.

"""Laboratório 6 — Credenciais e segredos do jeito certo.
Configure o token em .dlt/secrets.toml:
    [sources.github]
    api_token = "ghp_seu_token_pessoal_aqui"
Ou via variável de ambiente:
    export SOURCES__GITHUB__API_TOKEN="ghp_seu_token"
Sem token, a API do GitHub ainda funciona com rate limit mais baixo.
"""
import dlt
from dlt.sources.rest_api import rest_api_source
# Tenta ler o segredo; se não existir, roda sem auth (rate limit público)
try:
    token = dlt.secrets["sources.github.api_token"]
    auth = {"type": "bearer", "token": token}
    print("Usando autenticação bearer do secrets.toml / env.")
except Exception:
    auth = None
    print(
        "Aviso: sem token GitHub - rodando com rate limit público. "
        "Configure .dlt/secrets.toml ou SOURCES__GITHUB__API_TOKEN."
    )
client = {
    "base_url": "https://api.github.com/",
    "paginator": "header_link",
}
if auth:
    client["auth"] = auth
source = rest_api_source(
    {
        "client": client,
        "resource_defaults": {
            "write_disposition": "merge",
            "primary_key": "id",
        },
        "resources": [
            {
                "name": "issues",
                "endpoint": {
                    "path": "repos/duckdb/duckdb/issues",
                    "params": {"state": "open", "per_page": 30},
                },
            },
        ],
    }
)
pipeline = dlt.pipeline(
    pipeline_name="lab06_secrets",
    destination="duckdb",
    dataset_name="github",
)
load_info = pipeline.run(source)
print(load_info)
print("Issues:", len(pipeline.dataset().issues.df()))

Execução

# Sem token (ok para o lab; rate limit menor)
python lab06_secrets.py

# Com token
# edite .dlt/secrets.toml  OU  export SOURCES__GITHUB__API_TOKEN=...
python lab06_secrets.py

Critério de sucesso

• [ ] Pipeline termina com load bem-sucedido
• [ ] Imprime Issues: N com N > 0
• [ ] Sem token: mensagem de aviso; com token: “Usando autenticação bearer…”

Laboratório 7 — Trocando o destino: DuckDB → Postgres

Até aqui o destino foi DuckDB (local, rápido, descartável). A força do dlt: promover para produção mudando o destino — no nosso script, via flag.

Esse é o lab que, na minha experiência, converte céticos. Todo mundo que já reescreveu um pipeline inteiro porque ele “só rodava no meu banco” entende o valor de desenvolver barato e local e fazer deploy sem reescrever nada.

Pré-requisitos (só para --postgres)

uv pip install "dlt[postgres]"

Em .dlt/secrets.toml, descomente e preencha:

[destination.postgres.credentials]
database = "analytics"
username = "dlt_loader"
password = "senha_secreta"
host = "localhost"
port = 5432

Código (lab07_postgres.py)

"""Laboratório 7 — Trocando o destino: DuckDB → Postgres.Pré-requisitos:
  1. uv pip install "dlt[postgres]"
  2. Postgres rodando e acessível
  3. Credenciais em .dlt/secrets.toml:
     [destination.postgres.credentials]
     database = "analytics"
     username = "dlt_loader"
     password = "senha_secreta"
     host = "localhost"
     port = 5432
Por padrão este script usa DuckDB (sem Postgres).
Passe --postgres para apontar ao destino de produção.
"""
import argparse
import dlt
from dlt.sources.rest_api import rest_api_source

def main():
    parser = argparse.ArgumentParser(description="Lab 07 - troca de destino")
    parser.add_argument(
        "--postgres",
        action="store_true",
        help="Usa destination=postgres (requer credenciais em secrets.toml)",
    )
    args = parser.parse_args()
    destination = "postgres" if args.postgres else "duckdb"
    print(f"Destino: {destination}")
    source = rest_api_source(
        {
            "client": {
                "base_url": "https://jsonplaceholder.typicode.com/",
            },
            "resource_defaults": {
                "write_disposition": "merge",
                "primary_key": "id",
            },
            "resources": [
                "posts",
                "users",
            ],
        }
    )
    # A ÚNICA mudança entre local e produção é o argumento destination
    pipeline = dlt.pipeline(
        pipeline_name="lab07_prod",
        destination=destination,
        dataset_name="blog",
    )
    load_info = pipeline.run(source)
    print(load_info)
    if destination == "duckdb":
        print("Posts:", len(pipeline.dataset().posts.df()))
        print("Users:", len(pipeline.dataset().users.df()))

if __name__ == "__main__":
    main()

Execução

# Desenvolvimento local (sem Postgres)
python lab07_postgres.py
# Destino: duckdb
# Posts: 100
# Users: 10

# Produção (Postgres ligado + secrets)
python lab07_postgres.py --postgres

Fonte, recursos, PKs, merge: idênticos. Só muda destination. O schema é recriado no Postgres; merge vira INSERT ... ON CONFLICT.

O mesmo padrão vale para Snowflake, BigQuery, filesystem (Parquet no S3/MinIO), Iceberg/Delta, etc. E é aqui que o dlt encaixa na arquitetura de lakehouse open-source e self-hosted que defendo nesta série: ele é a camada EL na frente do lakehouse. Aponte o mesmo pipeline para o filesystem gravando Parquet particionado no seu MinIO, ou para uma tabela Delta/Iceberg, e você tem o bronze/raw alimentado; o dbt ou SQLMesh fazem o T por cima. O dlt é a peça que faltava entre "o dado existe em algum lugar" e "o dado está no meu object storage, pronto para consulta".

Critério de sucesso

• [ ] Sem flag: Destino: duckdb, Posts 100 / Users 10
• [ ] Com --postgres e credenciais: load no Postgres sem reescrever a fonte

Laboratório 8 — Agendamento e deploy

Um pipeline só vira útil quando roda sozinho. O dlt é “só Python”. Três caminhos comuns:

1. Cron — simples; pouca observabilidade.
2. Airflow / Kestra — produção séria (retries, alertas, painel).
3. Serverless / containers — estado no destino, não no disco local efêmero.

O ponto do item 3 conecta com tudo o que vimos: como o dlt guarda o estado no destino e não em disco local, ele roda sem sustos em ambientes efêmeros — um job Kubernetes que nasce e morre, uma Lambda, um Cloud Run. O filesystem sumir entre execuções não quebra o incremental, porque o marcador do cursor está no banco, não no contêiner.

Boas práticas: segredos por env; pipelines no Git; monitore load_info.

O que o repo entrega (lab08_airflow_sketch.py)

Não é um DAG instalável — é referência imprimível:

python lab08_airflow_sketch.py

Conteúdo essencial do arquivo:

"""Laboratório 8 — Esboço de deploy / agendamento.
O dlt é "só Python": roda em cron, Airflow, Kestra, containers, etc.
Este arquivo mostra um esboço de DAG Airflow e um entrypoint para cron.
NÃO é um DAG pronto para deploy - é referência do artigo.
"""
# crontab exemplo:
# 0 * * * * cd /path/lab-dlt && .venv/bin/python lab04_incremental.py
AIRFLOW_DAG_SKETCH = '''
from datetime import datetime
from airflow import DAG
from airflow.operators.python import PythonOperator

def run_dlt_pipeline():
    """Embrulha qualquer pipeline dlt numa task Airflow."""
    import dlt
    from dlt.sources.rest_api import rest_api_source
    source = rest_api_source({
        "client": {"base_url": "https://jsonplaceholder.typicode.com/"},
        "resource_defaults": {
            "write_disposition": "merge",
            "primary_key": "id",
        },
        "resources": ["posts", "users"],
    })
    pipeline = dlt.pipeline(
        pipeline_name="lab08_airflow",
        destination="duckdb",  # em prod: postgres / snowflake / etc.
        dataset_name="blog",
    )
    load_info = pipeline.run(source)
    print(load_info)

with DAG(
    dag_id="dlt_jsonplaceholder",
    start_date=datetime(2024, 1, 1),
    schedule="@hourly",
    catchup=False,
    tags=["dlt", "ingestao"],
) as dag:
    PythonOperator(
        task_id="run_pipeline",
        python_callable=run_dlt_pipeline,
    )
'''
if __name__ == "__main__":
    print("Lab 08 - referência de deploy")
    print()
    print("Cron:       python lab04_incremental.py  (via crontab)")
    print("Airflow:    veja AIRFLOW_DAG_SKETCH neste arquivo")
    print("Serverless: empacote num container; estado fica no destino")
    print()
    print("--- Esboço da DAG ---")
    print(AIRFLOW_DAG_SKETCH)

Como a série já cobriu Airflow em profundidade, a integração é natural: cada pipeline dlt vira uma task, e o estado incremental garante que reexecuções e retries não dupliquem dados. É a combinação que fecha uma das stacks de dados open-source mais sólidas que existem hoje: dlt (EL) + Airflow (orquestração) + dbt/SQLMesh (T).

Critério de sucesso

• [ ] Script imprime o esboço da DAG sem erro
• [ ] Você consegue imaginar o crontab apontando para lab04_incremental.py

Roteiro completo de teste (checklist)

Com o venv ativo e rede disponível:

source .venv/bin/activate

python lab01_basico.py
python lab02_rest.py
python lab03_merge.py && python lab03_merge.py   # Posts: 100 nas duas
python lab04_incremental.py && python lab04_incremental.py
python lab05_transform.py
python lab06_secrets.py
python lab07_postgres.py
python lab08_airflow_sketch.py

Validação deste repositório: todos os labs acima passaram com dlt 1.30.0.

Erros comuns e como depurar

Um material “do zero ao expert” precisa te preparar para quando algo dá errado — porque vai dar. Os tropeços mais frequentes com dlt são poucos e repetidos:

• Tabela duplicada. Você esqueceu do merge/primary_key e rodou o pipeline duas vezes com append (o padrão). Sintoma clássico do Lab 3. Solução: defina write_disposition: "merge" e uma primary_key que realmente identifique a linha.
• MissingDependencyException no load. Você escolheu um destino cujo extra não foi instalado — pediu duckdb sem dlt[duckdb], ou postgres sem dlt[postgres]. Instale o extra correspondente.
• Rate limit / 403 numa API. Rodou o Lab 4 ou 6 sem token e bateu no limite público. Configure o segredo (Lab 6) e caia no caminho autenticado.
• Incremental “não anda” ou traz tudo de novo. Ou o cursor_path aponta para um campo que não existe/não é ordenável, ou você misturou incremental com append e está duplicando a borda. Reveja o par cursor + merge.
• Colunas que aparecem “do nada”. Não é bug — é schema evolution fazendo o trabalho dela (intervalo conceitual acima). Se isso não pode acontecer naquela tabela, aplique um schema contract com freeze.

Para investigar sem abrir o banco na mão, o dlt tem um CLI de inspeção. Depois de rodar um pipeline, estes dois comandos resolvem a maioria das dúvidas:

# Estado geral do pipeline: destino, dataset, schemas, últimas cargas
dlt pipeline lab04_github info

# O trace da última execução: o que rodou em extract/normalize/load,
# tempos, contagens e onde falhou (se falhou)
dlt pipeline lab04_github trace

E, dentro do próprio script, o objeto load_info que os labs já imprimem é a primeira fonte de verdade: ele diz o destino, os pacotes carregados e sinaliza jobs que falharam. Não ignore esse print(load_info) — em produção, é o que você loga e monitora.

Quando usar (e quando NÃO usar) o dlt

Use dlt quando:

• Múltiplas fontes heterogêneas com schema que muda
• Carga incremental idempotente sem gerenciar estado na mão
• Pipelines versionáveis em Git, testáveis localmente, portáveis entre destinos
• Camada EL de lakehouse/warehouse self-hosted

Talvez não valha a pena quando:

• Carga única, fonte simples → pandas.read_csv().to_sql() basta
• Transformações pesadas de negócio → dbt/SQLMesh, não ingestão
• Já há plataforma gerenciada paga (Fivetran, Airbyte Cloud) e o custo não é problema

Onde o dlt se posiciona

Para situar a ferramenta no mapa da ingestão open-source e comercial:

A leitura resumida: o dlt ocupa o espaço de quem quer controle e código versionável sem operar uma plataforma inteira, e é especialmente atraente para quem se assustou com a conta da nuvem gerenciada e migrou para stack self-hosted. Não é a escolha certa para todo cenário — é a escolha certa para esse.

O elefante na sala: dlt na era dos agentes

Os 81.000 pipelines de janeiro de 2026, em boa parte gerados por agentes, não são coincidência. A forma declarativa — “declare a fonte, declare o destino, o resto eu resolvo” — é exatamente o tipo de interface que uma LLM preenche com alta taxa de acerto. Você descreve a API em linguagem natural, o agente gera a config rest_api_source, e na maioria das vezes funciona de primeira.

Para quem, como eu, está de olho em IA agêntica aplicada a engenharia de dados, o dlt é um caso de estudo do que vem por aí: ferramentas cuja superfície de configuração é simples e regular o suficiente para serem dirigidas por agentes, não só por humanos. A engenharia de dados está deixando de ser “escrever o pipeline” para virar “especificar o pipeline e revisar o que o agente gerou” — e note que a revisão só é confiável porque o resultado é código versionável em Git, diffável em PR, testável localmente. Um pipeline preso numa UI clica-clica não se revisa assim. O dlt é um dos primeiros exemplos práticos de uma ferramenta de dados desenhada, na prática, para esse mundo.

Glossário rápido

Para consulta, os termos que atravessam o artigo:

• EL / ELT — Extract-Load (o que o dlt faz) e a fase de Transform que vem depois, dentro do destino.
• Source / Resource — fonte (um sistema) e recurso (uma “tabela” dentro dela).
• Pipeline / Destination — o maquinário que move os dados e o lugar onde eles aterrissam.
• Write disposition — a estratégia de escrita: append, replace ou merge.
• Merge / upsert — atualizar-ou-inserir por chave primária; base da idempotência.
• Idempotência — rodar N vezes produz o mesmo resultado que rodar uma.
• Incremental / cursor — trazer só o que mudou desde a última execução, guiado por um campo (o cursor).
• Flattening — achatar objetos aninhados em colunas com separador __.
• Tabela-filha — tabela separada criada a partir de uma lista aninhada, ligada por _dlt_parent_id.
• Schema evolution — o dlt aceitar mudanças de schema automaticamente (modo evolve).
• Schema contract — regras (evolve/freeze/discard) para controlar essa evolução por entidade.
• Coluna variante (__v_) — coluna extra criada quando um valor não cabe no tipo já existente.
• Tabelas _dlt_ — tabelas de sistema (_dlt_loads, _dlt_pipeline_state, _dlt_version) onde o dlt guarda estado e trilha de auditoria.

Recapitulando

• Conceitos: source, resource, pipeline, destination; extract → normalize → load.
• Lab 0: uv + requirements.txt + .dlt/.
• Lab 1: Python → DuckDB sem DDL.
• Lab 2: REST declarativa + flattening.
• Intervalo: schema evolution e contracts.
• Lab 3: merge + PK = idempotência.
• Lab 4: incremental por cursor (joia da coroa).
• Lab 5: higiene na porta com processing_steps.
• Lab 6: secrets.toml / env, com fallback público no lab.
• Lab 7: mesma fonte, destino via --postgres.
• Lab 8: cron / Airflow / serverless como referência.

O dlt resolve um problema real com elegância desproporcional ao tamanho da biblioteca: mover dados de A para B, com schema e estado gerenciados, em Python versionável.

Próximo passo natural da série: fechar o ELT — pegar o que o dlt depositou e modelar com dbt ou SQLMesh.

Até a próxima!