Este repositório contém fluxos de captura e subida de dados no datalake da Prefeitura do Rio de Janeiro. O repositório é gerido pelo Escritório Municipal de Dados (EMD) e alimentado de forma colaborativa com as equipes de dados e tecnologia das Secretarias.
💜 Todo o código é desenvolvido em Python utilizando o software livre Prefect.
pip
miniconda
, virtualenv
ou similares)Este repositório contém parte do código sob a licença GPL-3.0 e parte sob uma licença EULA. Todo código sob a licença EULA terá um cabeçalho indicando que é proprietário. Consulte os respectivos tópicos em LICENÇA para os termos e condições de cada licença.
git clone https://github.com/prefeitura-rio/pipelines
Abrí-lo no seu editor de texto
No seu ambiente de desenvolvimento, instalar poetry para gerenciamento de dependências
pip3 install poetry
poetry install
pre-commit install
orgao/ # diretório raiz para o órgão
|-- projeto1/ # diretório de projeto
|-- |-- __init__.py # vazio
|-- |-- constants.py # valores constantes para o projeto
|-- |-- flows.py # declaração dos flows
|-- |-- schedules.py # declaração dos schedules
|-- |-- tasks.py # declaração das tasks
|-- |-- utils.py # funções auxiliares para o projeto
...
|-- __init__.py # importa todos os flows de todos os projetos
|-- constants.py # valores constantes para o órgão
|-- flows.py # declaração de flows genéricos do órgão
|-- schedules.py # declaração de schedules genéricos do órgão
|-- tasks.py # declaração de tasks genéricas do órgão
|-- utils.py # funções auxiliares para o órgão
orgao2/
...
utils/
|-- __init__.py
|-- flow1/
|-- |-- __init__.py
|-- |-- flows.py
|-- |-- tasks.py
|-- |-- utils.py
|-- flows.py # declaração de flows genéricos
|-- tasks.py # declaração de tasks genéricas
|-- utils.py # funções auxiliares
constants.py # valores constantes para todos os órgãos
O script manage.py
é responsável por criar e listar projetos desse repositório. Para usá-lo, no entanto, você deve instalar as dependências em requirements-cli.txt
:
pip3 install -r requirements-cli.txt
Você pode obter mais informações sobre os comandos com
python manage.py --help
O comando add-agency
permite que você adicione um novo órgão a partir do template padrão. Para fazê-lo, basta executar
python manage.py add-agency nome-do-orgao
Isso irá criar um novo diretório com o nome nome-do-orgao
em pipelines/
com o template padrão, já adaptado ao nome do órgão. O nome do órgão deve estar em snake case e deve ser único. Qualquer conflito com um projeto já existente será reportado.
Para listar os órgão existentes e nomes reservados, basta fazer
python manage.py list-projects
Em seguida, leia com anteção os comentários em cada um dos arquivos do seu projeto, de modo a evitar conflitos e erros. Links para a documentação do Prefect também encontram-se nos comentários.
Caso o órgão para o qual você desenvolverá um projeto já exista, basta fazer
python manage.py add-project nome-do-orgao nome-do-projeto
poetry add <package>
Requisitos do manage.py
estão em requirements-cli.txt
Requisitos para a Action de deployment estão em requirements-deploy.txt
Requisitos para testes estão em requirements-tests.txt
Escolha a pipeline que deseja executar (exemplo pipelines.rj_escritorio.template_pipeline.flows.flow
)
from pipelines.utils.utils import run_local
pipelines.rj_escritorio.template_pipeline.flows import flow
run_local(flow, parameters = {"param": "val"})
.env
na raiz
do projeto:GOOGLE_APPLICATION_CREDENTIALS=/path/to/credentials.json # Credenciais do Google Cloud
PREFECT__BACKEND=cloud
PREFECT__SERVER__HOST=https://prefect.dados.rio/api
PREFECT__SERVER__PORT=443
VAULT_ADDRESS=https://vault.dados.rio/
VAULT_TOKEN=<token> # Valor do token do órgão para o qual você está desenvolvendo. Caso não saiba o token, entre em contato.
source .env
Também, garanta que o arquivo $HOME/.prefect/auth.toml
exista e tenha um conteúdo semelhante a:
# This file is auto-generated and should not be manually edited
# Update the Prefect config or use the CLI to login instead
["prefect.dados.rio"]
api_key = "<sua-api-key>"
tenant_id = "<tenant-id>"
test.py
com a pipeline que deseja executar e adicione a função run_cloud
com os parâmetros necessários:from pipelines.utils import run_cloud
from pipelines.[secretaria].[pipeline].flows import flow # Complete com as infos da sua pipeline
run_cloud(
flow, # O flow que você deseja executar
labels=[
"example", # Label para identificar o agente que irá executar a pipeline (ex: rj-sme)
],
parameters = {
"param": "val", # Parâmetros que serão passados para a pipeline (opcional)
}
)
python test.py
A saída deve se assemelhar ao exemplo abaixo:
[2022-02-19 12:22:57-0300] INFO - prefect.GCS | Uploading xxxxxxxx-development/2022-02-19t15-22-57-694759-00-00 to datario-public
Flow URL: http://localhost:8080/default/flow/xxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
└── ID: xxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
└── Project: main
└── Labels: []
Run submitted, please check it at:
http://prefect-ui.prefect.svc.cluster.local:8080/flow-run/xxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx