feat: migra pipelines de dados do Prefect 0 para o Prefect 3

.github/scripts/deploy_flows.py

@@ -0,0 +1,172 @@
+"""
+Script de deploy de flows para o Prefect 3.
+
+Uso:
+  # Deploy de flows alterados em um PR (dev)
+  python deploy_flows.py --pool basedosdados-dev --branch feat/meu-flow --files pipelines/datasets/meu_dataset/flows.py
+
+  # Deploy de todos os flows (prod, ao mergear na main)
+  python deploy_flows.py --pool basedosdados --branch main --all
+"""
+
+import argparse
+import importlib.util
+import os
+import sys
+from pathlib import Path
+
+from prefect import Flow
+from prefect.runner.storage import GitRepository
+from prefect.schedules import Cron
+
+REPO_URL = "https://github.com/basedosdados/pipelines.git"
+
+
+def load_flows_from_file(file_path: str) -> dict[str, Flow]:
+    """
+    Importa dinamicamente um arquivo Python e retorna os flows Prefect 3 encontrados.
+    Arquivos que ainda usam Prefect 0.15.9 vão falhar na importação e serão pulados.
+    """
+    path = Path(file_path)
+    module_name = path.stem
+
+    spec = importlib.util.spec_from_file_location(module_name, file_path)
+    if spec is None or spec.loader is None:
+        print(f"  Pulando {file_path}: não foi possível carregar o spec.")
+        return {}
+
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[module_name] = module
+
+    try:
+        spec.loader.exec_module(module)
+    except ImportError as e:
+        print(
+            f"  Pulando {file_path}: ImportError (provavelmente Prefect 0.x) — {e}"
+        )
+        return {}
+    except Exception as e:
+        print(f"  Pulando {file_path}: erro ao carregar — {e}")
+        return {}
+
+    flows = {}
+    for name, obj in vars(module).items():
+        if isinstance(obj, Flow) and obj.fn.__code__.co_filename == str(
+            path.resolve()
+        ):
+            flows[name] = obj
+
+    return flows
+
+
+def deploy_flow(
+    flow: Flow,
+    flow_name: str,
+    file_path: str,
+    pool_name: str,
+    branch_name: str,
+) -> bool:
+    entrypoint = f"{file_path}:{flow_name}"
+    is_dev = "dev" in pool_name
+
+    schedules = getattr(flow, "deploy_schedules", None)
+    if is_dev:
+        schedules = None  # flows em dev não têm schedule
+    elif schedules:
+        # Convert dict {"cron": "...", "timezone": "..."} to Cron schedule objects
+        schedules = [
+            Cron(s["cron"], timezone=s.get("timezone", "UTC"))
+            if isinstance(s, dict)
+            else s
+            for s in schedules
+        ]
+
+    print(f"  Registrando {flow_name} → {entrypoint}")
+
+    try:
+        flow.from_source(
+            source=GitRepository(
+                url=REPO_URL,
+                branch=branch_name,
+            ),
+            entrypoint=entrypoint,
+        ).deploy(
+            name=flow_name,
+            work_pool_name=pool_name,
+            tags=["automated-deploy"],
+            schedules=schedules,
+            build=False,
+        )
+        status = (
+            "(sem schedule)"
+            if not schedules
+            else f"com schedules: {schedules}"
+        )
+        print(f"  ✓ {flow_name} registrado {status}")
+        return True
+    except Exception as e:
+        print(f"  ✗ Falha ao registrar {flow_name}: {e}")
+        return False
+
+
+def main():
+    parser = argparse.ArgumentParser(description="Deploy de flows Prefect 3")
+    parser.add_argument("--pool", required=True, help="Nome do Work Pool")
+    parser.add_argument(
+        "--branch", required=True, help="Branch do repositório"
+    )
+    parser.add_argument(
+        "--files", nargs="*", help="Arquivos específicos para deploy"
+    )
+    parser.add_argument(
+        "--all", action="store_true", help="Deploy de todos os flows"
+    )
+    args = parser.parse_args()
+
+    files_to_process = []
+
+    if args.all:
+        for root, _, files in os.walk("pipelines"):
+            for file in files:
+                if file.endswith(".py") and file != "__init__.py":
+                    files_to_process.append(os.path.join(root, file))
+    elif args.files:
+        files_to_process = args.files
+    else:
+        print("Nenhum arquivo especificado. Use --files ou --all.")
+        sys.exit(0)
+
+    print(f"\nWork Pool : {args.pool}")
+    print(f"Branch    : {args.branch}")
+    print(f"Arquivos  : {len(files_to_process)}\n")
+
+    success, skipped, failed = 0, 0, 0
+
+    for file_path in files_to_process:
+        if not os.path.exists(file_path):
+            continue
+
+        print(f"→ {file_path}")
+        flows = load_flows_from_file(file_path)
+
+        if not flows:
+            skipped += 1
+            continue
+
+        for name, flow_obj in flows.items():
+            ok = deploy_flow(flow_obj, name, file_path, args.pool, args.branch)
+            if ok:
+                success += 1
+            else:
+                failed += 1
+
+    print(
+        f"\nResultado: {success} registrados, {skipped} pulados, {failed} com erro"
+    )
+
+    if failed > 0:
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()

.github/workflows/build-docker-prefect3.yaml

@@ -0,0 +1,73 @@
+---
+name: Build Prefect 3 Docker image
+on:
+  push:
+    branches: [main]
+    paths:
+      - Dockerfile.prefect3
+      - entrypoint.sh
+      - pyproject.toml
+      - uv.lock
+      - packages.yml
+      - dbt_project.yml
+      - .github/workflows/build-docker-prefect3.yaml
+env:
+  GCP_PROJECT_ID: ${{ secrets.GCP_PROJECT_ID }}
+  IMAGE: gcr.io/${{ secrets.GCP_PROJECT_ID }}/pipelines
+  PREFECT3_API_URL: https://prefect3.basedosdados.org/api
+jobs:
+  build:
+    name: Build and push
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+      - name: Authenticate to Google Cloud
+        uses: google-github-actions/auth@v2
+        with:
+          credentials_json: ${{ secrets.GCP_SA_KEY }}
+          project_id: ${{ secrets.GCP_PROJECT_ID }}
+      - name: Set up Cloud SDK
+        uses: google-github-actions/setup-gcloud@v2
+      - name: Configure Docker for GCR
+        run: gcloud auth configure-docker --quiet
+      - name: Set image tag
+        id: tag
+        run: |
+          if [ "${{ github.ref_name }}" = "main" ]; then
+            echo "tag=latest" >> $GITHUB_OUTPUT
+          else
+            echo "tag=dev" >> $GITHUB_OUTPUT
+          fi
+      - name: Build and push
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          file: Dockerfile.prefect3
+          push: true
+          tags: ${{ env.IMAGE }}:${{ steps.tag.outputs.tag }}
+      - name: Update work pool image
+        env:
+          TAG: ${{ steps.tag.outputs.tag }}
+        run: |-
+          if [ "$TAG" = "latest" ]; then
+            POOL="basedosdados"
+          else
+            POOL="basedosdados-dev"
+          fi
+          curl -s "${{ env.PREFECT3_API_URL }}/work_pools/${POOL}" \
+            -H "Authorization: Bearer ${{ secrets.PREFECT3_AUTH_TOKEN }}" \
+            -o /tmp/wp.json
+          python3 -c "
+          import json
+          with open('/tmp/wp.json') as f:
+              wp = json.load(f)
+          wp['base_job_template']['job_configuration']['image'] = '${{ env.IMAGE }}:${TAG}'
+          with open('/tmp/wp-patch.json', 'w') as f:
+              json.dump({'base_job_template': wp['base_job_template']}, f)
+          "
+          curl -s -o /dev/null -w "Work pool '${POOL}' updated: HTTP %{http_code}\n" \
+            -X PATCH "${{ env.PREFECT3_API_URL }}/work_pools/${POOL}" \
+            -H "Authorization: Bearer ${{ secrets.PREFECT3_AUTH_TOKEN }}" \
+            -H "Content-Type: application/json" \
+            -d @/tmp/wp-patch.json

.github/workflows/build-docker.yaml

@@ -1,35 +1,7 @@
 ---
 name: Build Docker image
 on:
-  push:
-    branches: [main]
-    paths:
-      - .github/workflows/cd.yaml
-      - pipelines/**/*
-      - models/**/*
-      - macros/**/*
-      - tests-dbt/**/*
-      - pyproject.toml
-      - uv.lock
-      - Dockerfile
-      # DBT config files
-      - dbt_project.yml
-      - packages.yml
-  pull_request:
-    branches: [main]
-    paths:
-      - .github/workflows/cd-staging.yaml
-      - .github/workflows/build-docker.yaml
-      - pipelines/**/*
-      - models/**/*
-      - macros/**/*
-      - tests-dbt/**/*
-      - pyproject.toml
-      - uv.lock
-      - Dockerfile
-      # DBT config files
-      - dbt_project.yml
-      - packages.yml
+  workflow_dispatch:
 env:
   GCP_PROJECT_ID: ${{ secrets.GCP_PROJECT_ID }}
   GCP_SA_KEY: ${{ secrets.GCP_SA_KEY }}

.github/workflows/cd-prefect3-staging.yaml

@@ -0,0 +1,39 @@
+---
+name: cd-prefect3 (staging)
+on:
+  pull_request:
+    branches: [main, feat/prefect3]
+    types: [labeled, synchronize]
+    paths: [pipelines/**/*.py, .github/workflows/cd-prefect3-staging.yaml]
+jobs:
+  deploy-staging:
+    name: deploy flows (basedosdados-dev)
+    runs-on: ubuntu-latest
+    if: contains(github.event.pull_request.labels.*.name, 'deploy-flow')
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+      - name: Get changed flow files
+        id: changed-files
+        uses: tj-actions/changed-files@v47
+        with:
+          files: pipelines/**/*.py
+      - name: List changed files
+        if: steps.changed-files.outputs.any_changed == 'true'
+        run: echo "Flows alterados:${{ steps.changed-files.outputs.all_changed_files }}"
+      - name: Install uv
+        if: steps.changed-files.outputs.any_changed == 'true'
+        uses: astral-sh/setup-uv@v5
+      - name: Install dependencies
+        if: steps.changed-files.outputs.any_changed == 'true'
+        run: uv sync --locked --no-dev
+      - name: Deploy flows to basedosdados-dev
+        if: steps.changed-files.outputs.any_changed == 'true'
+        env:
+          PREFECT_API_URL: https://prefect3.basedosdados.org/api
+          PREFECT_API_KEY: ${{ secrets.PREFECT3_AUTH_TOKEN }}
+        run: |-
+          uv run python .github/scripts/deploy_flows.py \
+            --pool basedosdados-dev \
+            --branch ${{ github.head_ref }} \
+            --files ${{ steps.changed-files.outputs.all_changed_files }}

.github/workflows/cd-prefect3.yaml

@@ -0,0 +1,26 @@
+---
+name: cd-prefect3 (production)
+on:
+  push:
+    branches: [main]
+    paths: [pipelines/**/*.py, .github/workflows/cd-prefect3.yaml]
+jobs:
+  deploy-production:
+    name: deploy flows (basedosdados)
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+      - name: Install uv
+        uses: astral-sh/setup-uv@v5
+      - name: Install dependencies
+        run: uv sync --locked --no-dev
+      - name: Deploy all flows to basedosdados
+        env:
+          PREFECT_API_URL: https://prefect3.basedosdados.org/api
+          PREFECT_API_KEY: ${{ secrets.PREFECT3_AUTH_TOKEN }}
+        run: |-
+          uv run python .github/scripts/deploy_flows.py \
+            --pool basedosdados \
+            --branch ${{ github.ref_name }} \
+            --all

.github/workflows/cd-staging.yaml

@@ -1,23 +1,7 @@
 ---
 name: cd (staging)
 on:
-  pull_request:
-    branches: [main]
-    types: [labeled, opened, synchronize]
-    paths:
-      - .github/workflows/cd-staging.yaml
-      - pipelines/**/*
-      - pyproject.toml
-      - uv.lock
-      - Dockerfile
-      # dbt files
-      - models/**/*.sql
-      - models/**/*.yml
-      - macros/**/*
-      - tests-dbt/**/*
-      # dbt config files
-      - dbt_project.yml
-      - packages.yml
+  workflow_dispatch:
 env:
   GCP_PROJECT_ID: ${{ secrets.GCP_PROJECT_ID }}
   GCP_SA_KEY: ${{ secrets.GCP_SA_KEY }}

.github/workflows/cd.yaml

@@ -1,22 +1,7 @@
 ---
 name: cd (production)
 on:
-  push:
-    branches: [main]
-    paths:
-      - .github/workflows/cd.yaml
-      - pipelines/**/*
-      - pyproject.toml
-      - uv.lock
-      - Dockerfile
-      # dbt files
-      - models/**/*.sql
-      - models/**/*.yml
-      - macros/**/*
-      - tests-dbt/**/*
-      # dbt config files
-      - dbt_project.yml
-      - packages.yml
+  workflow_dispatch:
 env:
   GCP_PROJECT_ID: ${{ secrets.GCP_PROJECT_ID }}
   GCP_SA_KEY: ${{ secrets.GCP_SA_KEY }}