> ## Documentation Index > Fetch the complete documentation index at: https://docs.anyformat.ai/llms.txt > Use this file to discover all available pages before exploring further. # Daily CSV export > Bulk-export a workflow's runs for one local day as a flat CSV. The v2 "download as CSV" behavior, rebuilt on v3 with async concurrency, server-side date filtering and a client-side rate limit This recipe exports one local day of a workflow's runs as a flat CSV, one row per run. It runs no nodes of its own: it reads runs of any workflow that ends in an [Extract](/guides/nodes/extract) node, and those runs were billed when they ran.
One processed run
statusprocessed
invoice\_number{'INV-2026-9001'}
total\_amount{'4594.62'}
line\_items{'2 rows'}
{'↳ 134 runs in the 2026-07-14 window'}
→
One CSV row
{'run_id,packet_id,invoice_number,total_amount'}
{'aaaa…,06a8…,INV-2026-9001,4594.62'}
{'bbbb…,06a9…,INV-2026-1000,1810.50'}
{'cccc…,06aa…,INV-2026-1001,932.45'}
{'↳ scalars only; line_items is skipped'}
Nodes{'none of its own'} Credits0{' (it reads runs already billed; the Parse → Extract run it reads costs 60 per page)'} You get{' one row per run, one column per top-level scalar field'}
## When to use this * **v2 to v3 migration.** v2 exposed a per-workflow CSV download and v3 does not, so reproduce it in a few lines of SDK code. * **Daily reporting jobs.** Grab yesterday's runs on a cron and drop the CSV in a bucket, a warehouse or an email. * **Ad-hoc pulls.** Analysts asking for "everything we processed on 2026-07-14 in Madrid time". ## What it does 1. Convert the requested local day (**Europe/Madrid**) into a half-open UTC interval and hand it to `GET /v3/workflows/{id}/runs/` via the `document_packet_created_after` and `document_packet_created_before` query params. That is **server-side** filtering, one round-trip per page. 2. Fan out `GET /v3/runs/{run_id}/` in parallel to fetch inline results. 3. Emit one CSV row per run: `run_id`, `packet_id`, then one column per top-level scalar field. Nested list-of-object fields such as line items are skipped, as v2's CSV also flattened only scalars. The script assumes a linear `parse → extract` workflow with no `split` or `classify` branches. Concurrency and rate limiting, stdlib only: * `asyncio.Semaphore(CONCURRENCY)` caps in-flight requests. * A sliding-window **token bucket** at 55 req/min takes one `acquire()` per outbound HTTP call. That sits under the API's 60 req/min ceiling with headroom for a 429 retry, which honours `Retry-After`. Progress is logged to `stderr` at `INFO`: every request with its query params, the computed local to UTC window, page sizes, and per-run status. Read it to verify the timezone conversion at a glance. ## End-to-end ```python theme={null} """Export a workflow's runs for a given day (Europe/Madrid) as one flat CSV.""" import argparse import asyncio import csv import logging import os import sys from collections import deque from datetime import date, datetime, time, timedelta, timezone from zoneinfo import ZoneInfo import httpx from anyformat.sdk import AsyncClient, RateLimited BASE_URL = "https://api.anyformat.ai" MADRID = ZoneInfo("Europe/Madrid") RATE_PER_MIN = 55 # API cap is 60/min; leave headroom for the 429 retry CONCURRENCY = 8 logger = logging.getLogger("daily_csv_export") class TokenBucket: """Sliding-window '<= N requests per 60s' gate. Await before every HTTP call.""" def __init__(self, per_minute: int) -> None: self._n = per_minute self._times: deque[float] = deque() self._lock = asyncio.Lock() async def acquire(self) -> None: async with self._lock: loop = asyncio.get_running_loop() now = loop.time() while self._times and now - self._times[0] >= 60.0: self._times.popleft() if len(self._times) >= self._n: await asyncio.sleep(60.0 - (now - self._times[0])) self._times.append(loop.time()) def parse_day(text: str) -> date: """`today`, `yesterday`, or an ISO `YYYY-MM-DD`. Interpreted in Madrid.""" today = datetime.now(MADRID).date() if text == "today": return today if text == "yesterday": return today - timedelta(days=1) return date.fromisoformat(text) async def gated(bucket: TokenBucket, coro_factory): """One `bucket.acquire()` per outbound HTTP call, with one 429 retry.""" while True: await bucket.acquire() try: return await coro_factory() except RateLimited as e: logger.warning("429 rate limited; sleeping %.1fs", e.retry_after or 5.0) await asyncio.sleep(e.retry_after or 5.0) async def list_run_ids( http: httpx.AsyncClient, api_key: str, workflow_id: str, after_iso: str, before_iso: str, bucket: TokenBucket, ) -> list[str]: """Paginate GET /v3/workflows/{id}/runs/ with server-side date filtering. The SDK doesn't (yet) forward `document_packet_created_after` / `_before`, so we call the endpoint directly; the SDK covers the per-run fetch below.""" ids: list[str] = [] cursor: str | None = None headers = {"Authorization": f"Bearer {api_key}"} while True: params = { "limit": 100, "document_packet_created_after": after_iso, "document_packet_created_before": before_iso, } if cursor: params["cursor"] = cursor logger.info("GET /v3/workflows/%s/runs/ params=%s", workflow_id, params) async def _fetch() -> dict: r = await http.get( f"{BASE_URL}/v3/workflows/{workflow_id}/runs/", params=params, headers=headers, ) if r.status_code == 429: raise RateLimited(429, retry_after=float(r.headers.get("Retry-After") or 5.0)) r.raise_for_status() return r.json() body = await gated(bucket, _fetch) items = body.get("items") or [] cursor = body.get("next_cursor") logger.info(" -> %d runs on this page; next_cursor=%s", len(items), cursor) ids.extend(item["id"] for item in items) if not cursor: return ids async def fetch_run(client: AsyncClient, run_id: str, bucket: TokenBucket, sem: asyncio.Semaphore): async with sem: logger.info("GET /v3/runs/%s/", run_id) run = await gated(bucket, lambda: client.get_run(run_id)) logger.info(" -> status=%s", run.status) return run def scalar_cells(run) -> dict[str, object]: """Top-level scalar values from the (single) extraction. Skips nested list-of-object fields such as line items; v2's CSV did the same. Reads `result.raw["extractions"][0]["fields"]` rather than the typed `result.fields`: the typed view is silently empty on any wire drift, which shows up as a CSV with no columns.""" extractions = run.result.raw.get("extractions") or [] if not extractions: return {} return { name: val.get("value") for name, val in (extractions[0].get("fields") or {}).items() if isinstance(val, dict) and "value" in val } async def main() -> int: logging.basicConfig( level=logging.INFO, stream=sys.stderr, format="%(asctime)s %(levelname)s %(name)s: %(message)s", ) ap = argparse.ArgumentParser() ap.add_argument("--workflow", required=True) ap.add_argument("--day", default="yesterday", help="today | yesterday | YYYY-MM-DD") ap.add_argument("--out", default="runs.csv") args = ap.parse_args() day = parse_day(args.day) start_local = datetime.combine(day, time.min, MADRID) end_local = start_local + timedelta(days=1) start_utc = start_local.astimezone(timezone.utc) end_utc = end_local.astimezone(timezone.utc) # API accepts naive ISO 8601 as UTC. Drop the offset for a clean value. after_iso = start_utc.strftime("%Y-%m-%dT%H:%M:%S") before_iso = end_utc.strftime("%Y-%m-%dT%H:%M:%S") logger.info( "window day=%s Madrid=[%s, %s) UTC=[%s, %s)", day, start_local.isoformat(), end_local.isoformat(), after_iso, before_iso, ) api_key = os.environ["ANYFORMAT_API_KEY"] bucket = TokenBucket(RATE_PER_MIN) sem = asyncio.Semaphore(CONCURRENCY) async with httpx.AsyncClient(timeout=60.0) as http, AsyncClient(api_key=api_key) as client: run_ids = await list_run_ids(http, api_key, args.workflow, after_iso, before_iso, bucket) logger.info("total runs in window: %d", len(run_ids)) runs = await asyncio.gather( *(fetch_run(client, rid, bucket, sem) for rid in run_ids) ) processed = [r for r in runs if r.status == "processed" and r.result] logger.info("processed=%d (of %d fetched)", len(processed), len(runs)) # CSV: union of extracted-field columns across runs, first-seen order. prefix = ["run_id", "packet_id"] columns: list[str] = [] seen: set[str] = set(prefix) rows: list[dict] = [] for r in processed: cells = scalar_cells(r) for c in cells: if c not in seen: seen.add(c) columns.append(c) rows.append({"run_id": r.id, "packet_id": r.document_packet_id, **cells}) header = [*prefix, *columns] with open(args.out, "w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=header, extrasaction="ignore") w.writeheader() w.writerows(rows) logger.info("wrote %d rows -> %s", len(rows), args.out) return 0 if __name__ == "__main__": raise SystemExit(asyncio.run(main())) ``` ## Run it ```bash theme={null} export ANYFORMAT_API_KEY=af_... python daily_csv_export.py --workflow wf_123 --day yesterday --out yesterday.csv python daily_csv_export.py --workflow wf_123 --day 2026-07-14 --out 2026-07-14.csv ``` Expected stderr (abridged): ``` INFO daily_csv_export: window day=2026-07-14 Madrid=[2026-07-14T00:00:00+02:00, 2026-07-15T00:00:00+02:00) UTC=[2026-07-13T22:00:00, 2026-07-14T22:00:00) INFO daily_csv_export: GET /v3/workflows/wf_123/runs/ params={'limit': 100, 'document_packet_created_after': '2026-07-13T22:00:00', 'document_packet_created_before': '2026-07-14T22:00:00'} INFO daily_csv_export: -> 100 runs on this page; next_cursor=eyJ... INFO daily_csv_export: GET /v3/workflows/wf_123/runs/ params={'limit': 100, ..., 'cursor': 'eyJ...'} INFO daily_csv_export: -> 34 runs on this page; next_cursor=None INFO daily_csv_export: total runs in window: 134 INFO daily_csv_export: GET /v3/runs/aaaa.../ INFO daily_csv_export: -> status=processed ... INFO daily_csv_export: wrote 134 rows -> yesterday.csv ``` Check that the Madrid to UTC offset for the day looks right: `+02:00` in summer (CEST) with UTC bounds at `22:00`, `+01:00` in winter with `23:00`. Spain observes DST, so the exact UTC hours shift by month. This is a starting point, not a production template. ## When it goes wrong **The line items are missing from the CSV.** `scalar_cells` keeps only top-level scalars, because v2's CSV flattened only scalars too. An `object` field never becomes a column. Iterate the object list at `extraction["fields"]["line_items"]` (a `list[dict[str, {"value": ...}]]`) and emit one row per child, or write a second CSV. The same function reads `extractions[0]` alone, so a workflow with a Split or a Classify branch needs one row per extraction, not one per run. **The CSV has a header and no columns.** The typed `result.fields` view comes back silently empty on any wire drift, which lands as a CSV with `run_id` and `packet_id` and nothing else. `scalar_cells` reads `result.raw["extractions"][0]["fields"]` for that reason. Keep it reading the raw envelope. **You cannot tell which rows to trust.** `value` is one of three keys on every cell: `confidence` and `evidence` sit beside it. Append a `_confidence` column per field, or fold the evidence text in, and sort the export by the lowest confidence in the row. **The export stops partway with a 429.** The API ceiling is 60 requests a minute. `RATE_PER_MIN` is set to 55 so the single retry has room, and `gated()` honours `Retry-After`. Raise it only if your limit was raised, and keep it strictly under the ceiling. A run that fails anyway takes the whole export with it, because `asyncio.gather(...)` propagates the first exception: pass `return_exceptions=True` and log the per-run failures instead. **The day is off by an hour.** The window is built in Europe/Madrid and converted to UTC, and Spain observes DST, so the UTC bounds move between summer and winter. Read the logged window before you trust the file. For a range wider than a day, swap `parse_day` for a `--from` and `--to` pair and pass the wider range through the same `document_packet_created_{after,before}` params. ## Next steps Run lifecycle, statuses, and the results envelope How packets group files and relate to runs This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.