Orchestrate Brazil CVM Filings with Prefect
Need a Prefect CVM filings pipeline? Use Prefect for schedules and retries; use apicvm for resolve → list → download. This is pull-based monitoring — CVM does not push to your flow.
The problem
Teams already on Prefect want Brazilian regulatory PDFs in the same DAG graph as other market jobs. Scrapers inside tasks fail noisily. An HTTP API with document UUIDs fits Prefect's idempotent task model.
Minimal flow
import os, requests
from prefect import flow, task
BASE = os.environ["APICVM_URL"]
H = {"Authorization": f"Bearer {os.environ['APICVM_KEY']}"}
@task(retries=3, retry_delay_seconds=30)
def list_docs(ticker: str, year: int = 2024):
r = requests.get(
f"{BASE}/v1/documents",
headers=H,
params={"ticker": ticker, "type": "ITR", "year": year, "perPage": 50},
timeout=60,
)
r.raise_for_status()
return r.json()["data"]
@task
def download(doc_id: str, path: str):
r = requests.get(f"{BASE}/v1/documents/{doc_id}/file", headers=H, timeout=120)
r.raise_for_status()
open(path, "wb").write(r.content)
@flow
def itr_watch(tickers: list[str]):
for t in tickers:
for d in list_docs(t):
download(d["id"], f"/data/{t}/{d['id']}.pdf")
if __name__ == "__main__":
itr_watch(["PETR4", "VALE3", "ITUB4"])Scheduling
Deploy with a Prefect schedule (hourly/daily). Persist last-seen document IDs in your warehouse so re-runs are cheap. Compare with Airflow pipeline and GitHub Actions monitor.
Extraction tasks
If you need page markdown, fire POST /v1/document-text-extractions from a task and complete processing in a webhook receiver, not a busy-wait. There is no job-status polling endpoint in v1.
Current limitations
- Not real-time CVM push
- Rate limits per API key
- Empty lists mean ingestion lag, not "no filings exist"
Next steps
Ready to integrate?
Get an API key and start querying Brazilian CVM filings programmatically.