> ## Documentation Index
> Fetch the complete documentation index at: https://docs.dados.rio/llms.txt
> Use this file to discover all available pages before exploring further.

# Construindo uma Pipeline com Prefect 3

> Guia completo para criar pipelines de dados usando Prefect 3 na Prefeitura do Rio de Janeiro

# Construindo uma Pipeline com Prefect 3

Este guia explica como criar e configurar pipelines de dados usando Prefect 3 na Prefeitura do Rio de Janeiro, seguindo as melhores práticas e padrões estabelecidos pela equipe IplanRio.

## ✅ Pré-requisitos

### Acessos e Permissões

* Acesso ao Tailscale para conexão à rede interna da Prefeitura
* Permissão para acessar o Infisical (gerenciador de secrets)
* Acesso de leitura ao projeto BigQuery `rj-iplanrio`
* Permissões de colaborador no repositório GitHub

<Note>
  Caso não tenha acesso ao GitHub, BigQuery ou Infisical, solicite permissões no canal **#peça-permissão** do [Discord da IplanRio](https://discord.gg/Scr8HhzRuH).
</Note>

### Ambiente de Desenvolvimento

* Python 3.13+
* `uv` package manager
* Editor de código ([VSCode](https://code.visualstudio.com/) recomendado)
* [WSL 2](https://learn.microsoft.com/pt-br/windows/wsl/install) (usuários Windows)
* Conhecimento básico de Git e GitHub

## 🔧 Configuração Inicial

### 1. Clonar o Repositório

```bash theme={null}
git clone https://github.com/prefeitura-rio/prefect_rj_iplanrio.git
cd prefect_rj_iplanrio
```

### 2. Instalar Nix

O repositório utiliza Nix para gerenciar o ambiente de desenvolvimento.

```bash theme={null}
curl --proto '=https' --tlsv1.2 -L https://nixos.org/nix/install | sh -s -- --daemon
```

Verifique a instalação:

```bash theme={null}
which nix
```

### 3. Configurar direnv

O `direnv` gerencia automaticamente variáveis de ambiente do repositório.

```bash theme={null}
curl -sfL https://direnv.net/install.sh | bash
```

Adicione o hook ao seu shell:

```bash theme={null}
# Para Bash
eval "$(direnv hook bash)"

# Para Zsh, Oh My Zsh e outros shells, consulte: https://direnv.net/docs/hook.html
```

### 4. Instalar dependências

```bash theme={null}
uv sync
```

### 5. Configurar pre-commit hooks

```bash theme={null}
uv run pre-commit install --install-hooks
```

<Tip>
  Pre-commit hooks garantem qualidade do código antes do commit. Veja mais em [pre-commit.com](https://pre-commit.com/).
</Tip>

## 🚀 Boas Práticas de Desenvolvimento

### 1. Estrutura de Branch

```bash theme={null}
# Criar branch para sua pipeline
git checkout -b staging/nome-da-pipeline

# Exemplo para pipeline de saúde
git checkout -b staging/pipeline-dados-saude-sms
```

<Note>
  **IMPORTANTE**: Use sempre o prefixo `staging/` no nome da branch para que o CI/CD reconheça e processe sua pipeline automaticamente.
</Note>

### 2. Nomenclatura de Pipelines

Siga o padrão estabelecido:

```python theme={null}
# ✅ Correto: Nome descritivo e em snake_case
@flow(log_prints=True)
def rj_cvl__os_info():
    pass

# ❌ Evite: Nomes genéricos
@flow(log_prints=True)
def extracao_de_dados():
    pass
```

## 🧪 Criação de Nova Pipeline

### 1. Gerar Template com Cookiecutter

O repositório usa cookiecutter para criar pipelines padronizadas com Dockerfile, flow\.py, prefect.yaml e pyproject.toml.

```bash theme={null}
uvx cookiecutter templates --output-dir=pipelines
```

O comando solicitará informações sobre secretaria e pipeline:

```bash theme={null}
$ uvx cookiecutter templates --output-dir=pipelines
secretaria [sms]: sms
pipeline [dados_pacientes]: dados_saude_sms
```

<Tip>
  Use nomes descritivos que identifiquem a secretaria e o tipo de dados. Exemplo: `censo_escolar` (SME), `pacientes_upa` (SMS), `multas_transito` (SMTR).
</Tip>

### 2. Configurar flow\.py

Edite `flow.py` com as configurações específicas da sua pipeline:

```python theme={null}
# Exemplo: Pipeline de dados de pacientes SMS
@flow(log_prints=True)
def rj_sms__dados_pacientes_upa():
    # Configurações da base de dados
    db_database = "saude_producao"
    db_host = "10.20.30.40"
    db_port = 3306
    db_type = "mysql"
    
    # Dataset de destino no BigQuery
    dataset_id = "saude_sms"
    
    # Caminho dos secrets no Infisical
    infisical_secret_path = "/db-saude-sms"
```

<Warning>
  **Credenciais Obrigatórias**: Solicite ao IplanRio para adicionar usuário e senha do banco de origem no Infisical no caminho especificado em `infisical_secret_path`. Sem essas credenciais, a pipeline não conseguirá se conectar ao banco de dados.
</Warning>

**Parâmetros de Configuração:**

| Parâmetro               | Descrição                                              | Exemplo             |
| ----------------------- | ------------------------------------------------------ | ------------------- |
| `db_database`           | Nome do banco de dados de origem                       | `banco1`            |
| `db_host`               | Host do banco de dados de origem                       | `10.15.255.1`       |
| `db_port`               | Porta de conexão                                       | `3306`              |
| `db_type`               | Tipo do banco de dados                                 | `mysql`, `postgres` |
| `dataset_id`            | Dataset de destino no BigQuery (projeto `rj-iplanrio`) | `saude_sms`         |
| `infisical_secret_path` | Caminho dos secrets no Infisical                       | `/db-saude-sms`     |

### 3. Configurar prefect.yaml

Configure os schedules no arquivo `prefect.yaml` conforme suas necessidades:

#### Schedule Overwrite

Schedule overwrite substitui todos os dados da tabela a cada execução. Use para dados que devem refletir sempre o estado atual completo.

```yaml theme={null}
schedules:
  # Exemplo de schedule overwrite
  - interval: 86400 # executa a cada 24h
    anchor_date: "2022-01-01T01:00:00" # Data de inicio do schedule em UTC
    timezone: America/Sao_Paulo
    slug: dados_saude_sms_diario # slug do schedule
    parameters:
      table_id: pacientes_sms # Nome da tabela no BigQuery
      execute_query: |
        SELECT
          id_paciente,
          nome,
          data_nascimento,
          telefone
        FROM banco1.saude
```

<Tip>
  Use overwrite para tabelas dimensão ou cadastros que precisam refletir o estado atual completo, sem dados históricos.
</Tip>

#### Schedule Incremental

Schedule incremental adiciona apenas dados novos ou atualizados. Use para tabelas fato ou registros transacionais com alto volume.

```yaml theme={null}
schedules:
  # Exemplo de schedule incremental
  - interval: 86400
    anchor_date: "2022-01-01T02:32:00"
    timezone: America/Sao_Paulo
    slug: dados_saude_sms_incremental
    parameters:
      table_id: pacientes_sms_incremental
      dump_mode: append
      partition_columns: data_atualizacao # Coluna de particionamento
      partition_date_format: "%Y-%m-%d" # Formato da data de particionamento
      break_query_frequency: day # Frequencia de particionamento
      break_query_start: current_day # Data de inicio do particionamento
      break_query_end: current_day # Data de fim do particionamento
      execute_query: |
        SELECT
          id_paciente,
          nome,
          data_nascimento,
          telefone,
          data_atualizacao
        FROM banco1.saude
```

<Tip>
  Use incremental para dados que crescem continuamente (transações, eventos, logs). O particionamento otimiza queries e reduz custos no BigQuery.
</Tip>

**Parâmetros de Schedule:**

| Parâmetro               | Descrição                             | Valores                  |
| ----------------------- | ------------------------------------- | ------------------------ |
| `interval`              | Intervalo entre execuções em segundos | `86400` (24h)            |
| `anchor_date`           | Data de início do schedule em UTC     | `2022-01-01T01:00:00`    |
| `timezone`              | Fuso horário                          | `America/Sao_Paulo`      |
| `slug`                  | Identificador único do schedule       | `dados_saude_sms_diario` |
| `table_id`              | Nome da tabela no BigQuery            | `pacientes_sms`          |
| `dump_mode`             | Modo de inserção                      | `append`, `overwrite`    |
| `partition_columns`     | Coluna(s) para particionamento        | `data_atualizacao`       |
| `partition_date_format` | Formato da data de particionamento    | `%Y-%m-%d`               |
| `break_query_frequency` | Frequência de particionamento         | `day`, `month`, `year`   |
| `break_query_start/end` | Período de processamento              | `current_day`            |

### 4. Commit e Push

```bash theme={null}
# Adicionar arquivos
git add pipelines/sua_secretaria/sua_pipeline/

# Commit com mensagem descritiva
git commit -m "feat: adiciona pipeline de dados de saúde SMS

- Cria pipeline para extração de dados SMS
- Configura schedule diário incremental
- Adiciona particionamento por data"

# Push para branch
git push origin staging/sua-pipeline
```

<Tip>
  Use [Conventional Commits](https://www.conventionalcommits.org/) para mensagens padronizadas: `feat:`, `fix:`, `docs:`, `refactor:`.
</Tip>

### 5. Criar Pull Request

1. **Criar PR** no GitHub para a branch `staging/sua-pipeline`
2. **Descrever mudanças**: Explique o objetivo da pipeline e impacto no projeto
3. **Solicitar review** da equipe IplanRio
4. **Aguardar CI/CD**: Todos os testes devem passar
5. **Testar em Staging**: Valide a pipeline no ambiente de desenvolvimento após deploy
6. **Merge**: Após aprovação, faça merge para `main`

#### Workflow de CI/CD Automático

O repositório utiliza GitHub Actions para automatizar build, deploy e publicação de pipelines.

**🚀 Deploy Automático**

O sistema possui workflows separados para staging e produção:

| Ambiente | Workflow                            | Trigger             | Monitora       |
| -------- | ----------------------------------- | ------------------- | -------------- |
| Staging  | `deploy-prefect-flows-staging.yaml` | Push em `staging/*` | `pipelines/**` |
| Produção | `deploy-prefect-flows-prod.yaml`    | Push em `master`    | `pipelines/**` |

**🔧 Processo de Deploy**

Ambos os workflows executam:

1. Checkout do código-fonte
2. Login no GitHub Container Registry (ghcr.io)
3. Instalação de dependências Python com `uv`
4. Execução do script `.github/scripts/deploy_prefect_flows.py`
   * Deploy automático de todos os flows em `pipelines/*/prefect.yaml`
   * Falhas interrompem o workflow e registram erro nos logs

**🐳 Build da Imagem Base**

Workflow `build-and-push-root-dockerfile.yaml`:

* **Trigger**: Alterações no Dockerfile raiz ou push em `master`
* **Processo**: Build e publicação em `ghcr.io/${{ github.repository }}:latest`

<Note>
  Staging permite testar pipelines antes de produção. Após deploy em staging funcionar, teste no ambiente de desenvolvimento. Apenas após merge em `master` as pipelines são deployadas em produção.
</Note>

<Warning>
  Se algum deploy falhar, o workflow será interrompido. Corrija os problemas antes de tentar novamente.
</Warning>

**📊 Monitoramento**

* Acompanhe progresso na aba **Actions** do GitHub
* Verifique logs para identificar erros
* Aguarde conclusão antes de solicitar review
* Falhas requerem novo commit para re-executar CI/CD

## 🔧 Troubleshooting

### Qual work-pool utilizar:

A escolha do **work-pool** depende de onde a pipeline será executada e dos recursos que ela precisa acessar.

| Work-pool        | Quando utilizar                                                                                                                                                                                                                |
| ---------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
| **default-pool** | Utilize sempre que a pipeline **não precisar acessar recursos internos da Iplanrio** e **não pertencer ao projeto Táxi.Rio**.                                                                                                  |
| **k3s-pool**     | Pool padrão para a maioria das pipelines. Utilize para pipelines que precisam acessar recursos internos da **Iplanrio**, como bancos de dados, APIs, serviços ou outros recursos disponíveis apenas na infraestrutura interna. |
| **taxirio-pool** | Utilize exclusivamente para pipelines do projeto **Táxi.Rio**.                                                                                                                                                                 |

### Erro de Conexão com Banco de Dados:

* Verifique credenciais no Infisical no caminho `infisical_secret_path`
* Confirme acesso ao host e porta via Tailscale
* Teste conexão manualmente com ferramenta como `mysql-client` ou `psql`

### Falha no Deploy:

* Verifique logs na aba **Actions** do GitHub
* Confirme que todos os arquivos foram commitados (`flow.py`, `prefect.yaml`, `Dockerfile`)
* Valide sintaxe YAML em [yamllint.com](https://www.yamllint.com/)

### Pipeline Não Executa no Schedule:

* Verifique se `anchor_date` está no passado (não no futuro)
* Confirme `timezone: America/Sao_Paulo`
* Valide `interval` em segundos (86400 = 24 horas)

### Dados Não Aparecem no BigQuery:

* Confirme `dataset_id` correto no projeto `rj-iplanrio`
* Verifique se a query retorna dados executando-a manualmente
* Valide permissões de escrita no BigQuery (solicite ao IplanRio se necessário)

## 📚 Recursos Adicionais

* [Migração de Pipeline](https://iplan-rio.mintlify.app/data-lake/prefect/migracao-de-pipeline)
* [Repositório GitHub](https://github.com/prefeitura-rio/prefect_rj_iplanrio)
* [Documentação Prefect](https://docs.prefect.io/)

***