#!/usr/bin/env python3 """Push FXMacroData history into an Airtable synced table through the Sync API. Airtable's Sync API turns a CSV into a read-only synced table that Airtable keeps in step with each upload. This script reads one FXMacroData dataset, writes it as CSV, and posts it to your table's Sync API endpoint. Run it on a schedule (cron, Task Scheduler, a CI job) to keep the table current. Setup in Airtable (Business or Enterprise Scale plan, creator permission): "+ Add or import" > Sync API > Set up. Create a personal access token with the ``data.records:write`` and ``schema.bases:write`` scopes for that base, run this script once with ``--sync-url`` set to the endpoint in the cURL example Airtable shows, then pick field types and choose "Key" as the unique ID. Leave the deletion setting on "delete records not in the latest request" so the table mirrors the most recent upload. export FXMD_API_KEY=... # optional for USD indicators, calendar and COT export AIRTABLE_TOKEN=pat... python fxmd_airtable_sync.py indicator --currency usd --indicator inflation \\ --sync-url https://api.airtable.com/v0/appXXXX/tblXXXX/sync/XXXX Only the Python standard library is used. Docs: https://fxmacrodata.com/documentation/airtable?utm_source=github&utm_medium=referral&utm_campaign=airtable&utm_content=sync-api """ from __future__ import annotations import argparse import csv import io import json import os import sys import time import urllib.error import urllib.parse import urllib.request from datetime import datetime, timezone from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence API_ORIGIN = "https://api.fxmacrodata.com" USER_AGENT = "fxmacrodata-airtable-sync/1.0" PAGE_LIMIT = 100 # Airtable Sync API limits: 10,000 rows, 500 columns, 2 MB per request. MAX_SYNC_ROWS = 10_000 MAX_SYNC_BYTES = 2 * 1024 * 1024 SYNC_URL_PREFIX = "https://api.airtable.com/v0/" class SyncError(Exception): """A failure the user can act on; printed without a traceback.""" def _epoch_iso(value: Any) -> str: if isinstance(value, bool) or not isinstance(value, (int, float)): return "" return datetime.fromtimestamp(value, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def _num(value: Any) -> str: if isinstance(value, bool) or not isinstance(value, (int, float)): return "" if value != value or value in (float("inf"), float("-inf")): return "" return repr(value) if isinstance(value, float) else str(value) def _text(value: Any) -> str: return "" if value is None else str(value).strip() # Each dataset: API path, the CSV columns, and how one API row becomes one CSV # row. "Key" is unique per row; choose it as the unique ID when Airtable asks. DATASETS: Dict[str, Dict[str, Any]] = { "indicator": { "requires": ("currency", "indicator"), "free": True, "path": lambda a: f"/v1/announcements/{a.currency.lower()}/{a.indicator.lower()}", "query": {"fields": "compact"}, "paged": True, "columns": ["Key", "Date", "Value", "Previous", "Change", "Released at (UTC)", "Currency", "Indicator"], "row": lambda r, a: [ f"{a.currency.upper()} {a.indicator.lower()} {_text(r.get('date'))}", _text(r.get("date")), _num(r.get("val")), _num(r.get("previous_value")), _num(r.get("change_from_previous")), _epoch_iso(r.get("announcement_datetime")), a.currency.upper(), a.indicator.lower(), ], }, "forex": { "requires": ("base", "quote"), "free": False, "path": lambda a: f"/v1/forex/{a.base.lower()}/{a.quote.lower()}", "query": {}, "paged": True, "columns": ["Key", "Date", "Rate", "Open", "High", "Low", "Close", "Pair"], "row": lambda r, a: [ f"{a.base.upper()}{a.quote.upper()} {_text(r.get('date'))}", _text(r.get("date")), _num(r.get("val")), _num(r.get("open")), _num(r.get("high")), _num(r.get("low")), _num(r.get("close")), f"{a.base.upper()}/{a.quote.upper()}", ], }, "commodity": { "requires": ("commodity",), "free": False, "path": lambda a: f"/v1/commodities/{a.commodity.lower()}", "query": {}, "paged": True, "columns": ["Key", "Date", "Value", "Commodity", "Source"], "row": lambda r, a: [ f"{a.commodity.lower()} {_text(r.get('date'))}", _text(r.get("date")), _num(r.get("val")), a.commodity.lower(), _text(r.get("source")), ], }, "calendar": { "requires": ("currency",), "free": True, "path": lambda a: f"/v1/calendar/{a.currency.lower()}", "query": {}, "paged": False, "columns": ["Key", "Release", "Name", "Scheduled at (UTC)", "Reference period", "Importance", "Currency"], "row": lambda r, a: [ _text(r.get("calendar_event_id")) or f"{a.currency.lower()}_{_text(r.get('release'))}_{_text(r.get('announcement_datetime'))}", _text(r.get("release")), _text(r.get("name")), _epoch_iso(r.get("announcement_datetime")), _text(r.get("reference_period")), _text(r.get("event_importance")), a.currency.upper(), ], }, "cot": { "requires": ("currency",), "free": True, "path": lambda a: f"/v1/cot/{a.currency.lower()}", "query": {}, "paged": True, "columns": [ "Key", "Date", "Open interest", "Non-commercial net", "Commercial net", "Leveraged funds net", "Asset manager net", "Dealer net", "Released at (UTC)", ], "row": lambda r, a: [ f"{a.currency.upper()} COT {_text(r.get('date'))}", _text(r.get("date")), _num(r.get("open_interest")), _num(r.get("noncommercial_net")), _num(r.get("commercial_net")), _num(r.get("leveraged_funds_net")), _num(r.get("asset_manager_net")), _num(r.get("dealer_net")), _epoch_iso(r.get("announcement_datetime")), ], }, } Opener = Callable[[urllib.request.Request, float], Any] def _default_opener(request: urllib.request.Request, timeout: float) -> Any: # Only two https hosts are ever opened: API_ORIGIN, and a sync URL that # parse_args has already checked starts with SYNC_URL_PREFIX. return urllib.request.urlopen(request, timeout=timeout) # nosec B310 # noqa: S310 def _read_json(response: Any) -> Any: raw = response.read() try: return json.loads(raw.decode("utf-8") or "null") except (UnicodeDecodeError, json.JSONDecodeError): return None def _http_error_detail(err: urllib.error.HTTPError) -> str: try: body = json.loads(err.read().decode("utf-8") or "null") except Exception: # noqa: BLE001 - best-effort detail only body = None if isinstance(body, dict): for key in ("detail", "error", "message"): value = body.get(key) if isinstance(value, str): return value if isinstance(value, dict) and isinstance(value.get("message"), str): return value["message"] return err.reason if isinstance(err.reason, str) else "" def fetch_rows( args: argparse.Namespace, api_key: Optional[str], *, opener: Opener = _default_opener, timeout: float = 30.0, ) -> List[Dict[str, Any]]: spec = DATASETS[args.dataset] headers = {"Accept": "application/json", "User-Agent": USER_AGENT} if api_key: # Header, not query string: keys stay out of proxy and server logs. headers["X-API-Key"] = api_key rows: List[Dict[str, Any]] = [] offset = 0 for _ in range(1000): query = dict(spec["query"]) if args.start_date: query["start_date"] = args.start_date if args.end_date: query["end_date"] = args.end_date if spec["paged"]: query.update({"limit": PAGE_LIMIT, "offset": offset}) url = API_ORIGIN + spec["path"](args) if query: url += "?" + urllib.parse.urlencode(query) request = urllib.request.Request(url, headers=headers, method="GET") try: response = opener(request, timeout) except urllib.error.HTTPError as err: detail = _http_error_detail(err) hint = "" if err.code in (401, 403) and not api_key: hint = " Set FXMD_API_KEY; only USD indicators, calendar and COT are free." raise SyncError(f"FXMacroData HTTP {err.code}: {detail}{hint}".rstrip()) from None except urllib.error.URLError as err: raise SyncError(f"Could not reach FXMacroData: {err.reason}") from None body = _read_json(response) data = body.get("data") if isinstance(body, dict) else None if not isinstance(data, list): raise SyncError("FXMacroData returned an unexpected response.") for row in data: if len(rows) >= args.max_rows: break if isinstance(row, dict): rows.append(row) pagination = body.get("pagination") or {} if not spec["paged"] or not data or len(rows) >= args.max_rows or pagination.get("has_more") is not True: break next_offset = pagination.get("next_offset") offset = next_offset if isinstance(next_offset, int) else offset + len(data) return rows def build_csv(args: argparse.Namespace, rows: Iterable[Dict[str, Any]]) -> bytes: spec = DATASETS[args.dataset] buffer = io.StringIO() writer = csv.writer(buffer, lineterminator="\n") writer.writerow(spec["columns"]) seen = set() count = 0 for row in rows: values = spec["row"](row, args) if not values[0] or values[0] in seen: continue seen.add(values[0]) writer.writerow(values) count += 1 if count >= MAX_SYNC_ROWS: break payload = buffer.getvalue().encode("utf-8") if len(payload) > MAX_SYNC_BYTES: raise SyncError( f"The CSV is {len(payload):,} bytes; Airtable accepts at most 2 MB. " "Use --max-rows or --start-date to send less." ) return payload def push_csv( sync_url: str, token: str, payload: bytes, *, opener: Opener = _default_opener, timeout: float = 60.0, ) -> Any: request = urllib.request.Request( sync_url, data=payload, headers={ "Authorization": f"Bearer {token}", "Content-Type": "text/csv", "User-Agent": USER_AGENT, }, method="POST", ) try: return _read_json(opener(request, timeout)) except urllib.error.HTTPError as err: detail = _http_error_detail(err) if err.code == 429: detail = (detail + " Airtable allows 20 Sync API requests per 5 minutes per base.").strip() raise SyncError(f"Airtable HTTP {err.code}: {detail}".rstrip()) from None except urllib.error.URLError as err: raise SyncError(f"Could not reach Airtable: {err.reason}") from None def _valid_date(value: str) -> str: try: if len(value) != 10: raise ValueError(value) time.strptime(value, "%Y-%m-%d") except ValueError: raise argparse.ArgumentTypeError("dates must be YYYY-MM-DD") from None return value def _max_rows(value: str) -> int: number = int(value) if not 1 <= number <= MAX_SYNC_ROWS: raise argparse.ArgumentTypeError(f"must be between 1 and {MAX_SYNC_ROWS}") return number def parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace: parser = argparse.ArgumentParser( description="Push FXMacroData history into an Airtable Sync API table.", ) parser.add_argument("dataset", choices=sorted(DATASETS)) parser.add_argument("--currency") parser.add_argument("--indicator") parser.add_argument("--base") parser.add_argument("--quote") parser.add_argument("--commodity") parser.add_argument("--start-date", type=_valid_date) parser.add_argument("--end-date", type=_valid_date) parser.add_argument("--max-rows", type=_max_rows, default=5000) parser.add_argument( "--sync-url", default=os.environ.get("AIRTABLE_SYNC_URL"), help="The table's Sync API endpoint (or AIRTABLE_SYNC_URL).", ) parser.add_argument("--dry-run", action="store_true", help="Print the CSV instead of uploading it.") args = parser.parse_args(argv) missing = [name for name in DATASETS[args.dataset]["requires"] if not getattr(args, name)] if missing: parser.error(f"{args.dataset} needs --{' --'.join(missing)}") if not args.dry_run: if not args.sync_url: parser.error("--sync-url (or AIRTABLE_SYNC_URL) is required unless --dry-run is set") if not args.sync_url.startswith(SYNC_URL_PREFIX) or "/sync/" not in args.sync_url: parser.error("--sync-url must be an https://api.airtable.com/v0/.../sync/... endpoint") return args def main(argv: Optional[Sequence[str]] = None, *, opener: Opener = _default_opener) -> int: args = parse_args(argv) api_key = (os.environ.get("FXMD_API_KEY") or "").strip() or None token = (os.environ.get("AIRTABLE_TOKEN") or "").strip() if not args.dry_run and not token: print( "Set AIRTABLE_TOKEN to a personal access token with data.records:write " "and schema.bases:write.", file=sys.stderr, ) return 2 try: rows = fetch_rows(args, api_key, opener=opener) payload = build_csv(args, rows) if args.dry_run: sys.stdout.write(payload.decode("utf-8")) return 0 result = push_csv(args.sync_url, token, payload, opener=opener) except SyncError as err: print(str(err), file=sys.stderr) return 1 sent = max(payload.count(b"\n") - 1, 0) print(f"Sent {sent} rows to Airtable.") if isinstance(result, dict) and result.get("success") is False: print(json.dumps(result), file=sys.stderr) return 1 return 0 if __name__ == "__main__": sys.exit(main())