Orchestrate Brazil CVM Filings with Dagster

The problem

Data platforms that keep Brazilian filing inventories fresh need:

• Idempotent document UUID keys
• Scheduled coverage diffs (what is new since last run)
• Clear separation between metadata assets and PDF/blob assets

Portal scrapers fail quietly in CI. A fixed /v1 contract is easier to soft-fail and alert on.

Asset sketch

import os, requests
from dagster import asset, AssetExecutionContext

BASE = os.environ["APICVM_URL"]
H = {"Authorization": f"Bearer {os.environ['APICVM_KEY']}"}

@asset
def petr4_dfp_2024(context: AssetExecutionContext) -> list[dict]:
    r = requests.get(
        f"{BASE}/v1/documents",
        headers=H,
        params={"ticker": "PETR4", "type": "DFP", "year": 2024, "perPage": 50},
        timeout=60,
    )
    r.raise_for_status()
    rows = r.json()["data"]
    context.log.info("found %s documents", len(rows))
    return rows

Download step

Materialize PDFs with GET /v1/documents/:id/file into object storage keyed by document UUID. File download does not consume extraction credits.

For page markdown, enqueue POST /v1/document-text-extractions with an HTTPS callback (Pro) and treat the callback as a separate sensor/ingest path — there is no job-status poll endpoint.

Scheduling

Use a Dagster schedule or sensor that re-lists ITR for a watchlist each trading week. Empty lists are corpus gaps, not always API failures — log and continue.

Complementary: Airflow pipeline, Prefect pipeline.

Current limitations

• Pull-based only; no CVM push feed in v1
• Async extraction via callback only
• Rate limits per API key (X-RateLimit-*)

Next steps

• Read the API docs
• Get an API key
• GitHub Actions monitor
• Paginate and filter

Related guides

• dbt staging
• Handle errors and rate limits