from __future__ import annotations import json import subprocess import sys from datetime import UTC, datetime from pathlib import Path from typing import Any from .config import LoadConfig def write_event(config: LoadConfig, manifest: dict, event: dict[str, Any]) -> dict[str, str]: """Append one event to the local run log. The log is uploaded once at the end of the run by ``upload_runtime_log`` — not per event — so an N-table run does not re-send a steadily growing object N times (O(N^2) bytes).""" run_date = manifest["run_date"] run_id = manifest["run_id"] record = { "event_schema_version": 1, "source_name": manifest["source_name"], "dataset": manifest.get("dataset", ""), "run_date": run_date, "run_id": run_id, "created_at": datetime.now(UTC).isoformat(), **event, } output_path = runtime_log_path(config, run_date, run_id) output_path.parent.mkdir(parents=True, exist_ok=True) with output_path.open("a") as file: file.write(json.dumps(record, sort_keys=True) + "\n") return {"local_path": str(output_path), "gcs_uri": manifest["runtime_log_uri"]} def upload_runtime_log(config: LoadConfig, manifest: dict) -> dict[str, str] | None: """Upload the complete run log to GCS once, after the run. Best-effort: a failed upload must not abort an otherwise-successful load — the raw data has already landed and the log remains on local disk. Returns the destination (with an ``uploaded`` flag), or None if there is nothing to upload. """ output_path = runtime_log_path(config, manifest["run_date"], manifest["run_id"]) if not output_path.exists(): return None gcs_uri = manifest["runtime_log_uri"] try: subprocess.run( ["gcloud", "storage", "cp", str(output_path), gcs_uri], check=True, capture_output=True, text=True, ) except (subprocess.CalledProcessError, OSError) as error: print( json.dumps({"runtime_log_upload_failed": gcs_uri, "error": str(error)}), file=sys.stderr, ) return {"local_path": str(output_path), "gcs_uri": gcs_uri, "uploaded": "false"} return {"local_path": str(output_path), "gcs_uri": gcs_uri, "uploaded": "true"} def runtime_log_path(config: LoadConfig, run_date: str, run_id: str) -> Path: return config.root / "runs" / run_date / run_id / "events.jsonl"