> ## 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
## When to use this
* **v2 → v3 migration** — v2 exposed a per-workflow CSV download; v3 doesn't, so
reproduce it in a few lines of SDK code.
* **Daily reporting jobs** — grab yesterday's runs on a cron, drop the CSV in a
bucket / warehouse / 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` / `document_packet_created_before` query
params — **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 (e.g. line items)
are skipped — v2's CSV also flattened only scalars. Assumes a linear
`parse → extract` workflow (no `split` / `classify` branches).
Concurrency + rate limiting, stdlib only:
* `asyncio.Semaphore(CONCURRENCY)` — in-flight request cap.
* A sliding-window **token bucket** at 55 req/min — one `acquire()` per
outbound HTTP call, 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↔UTC window, page sizes, and per-run status — so you can
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 (e.g. 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-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.
## Adapting it
This is a starting point, not a production template. Common tweaks:
* **Nested rows** — need line items? For each processed run, 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.
* **Confidence / evidence columns** — each cell dict carries `confidence` and
`evidence` alongside `value`; append `_confidence` columns or fold
evidence text in as needed.
* **Windowing** — swap `parse_day` for a `--from` / `--to` pair and pass the
wider range as the same `document_packet_created_{after,before}` params.
* **Higher throughput** — bump `RATE_PER_MIN` if you have a raised limit;
keep it strictly under the ceiling so the retry has room.
* **Failure isolation** — swap `asyncio.gather(...)` for
`asyncio.gather(..., return_exceptions=True)` and log per-run failures
instead of aborting the whole export.
## Next steps
Run lifecycle, statuses, and the results envelope
How packets group files and relate to runs