Supply Chain & Retail Solutions API guide

"csv_path":    "./orders/",
"object_name": "acme_orders_v1",
"file_glob":   "*.csv.gz",     # use "*.csv" if you disabled gzip with COMPRESSION = NONE
"csv_path":    "./orders/",
"object_name": "acme_orders_v1",
"file_glob":   "*.csv.gz",     # use "*.csv" if you disabled gzip with COMPRESSION = NONE

"csv_path":    "./orders/",
"object_name": "acme_orders_v1",
"file_glob":   "*.csv",        # default; use "*" if you kept Redshift's extensionless parts, or "*.gz" if you enabled GZIP
"csv_path":    "./orders/",
"object_name": "acme_orders_v1",
"file_glob":   "*.csv",        # default; use "*" if you kept Redshift's extensionless parts, or "*.gz" if you enabled GZIP

#!/usr/bin/env python3
"""
Bulk-ingest historical CSV data into the Peak Data Ingestion v2 API.

- One run loads one table from a CSV file or a folder of CSV part-files.
- Streams the source row-by-row (no full in-memory load).
- Dynamically sizes batches to <=2000 rows AND <=1 MB serialized.
- Concurrent workers share a token-bucket rate limiter.
- Exponential backoff on 5xx / 429; never retries 4xx.
- Checkpoint lets multi-hour runs resume cleanly after a crash.
- Failed rows from 207 responses written to a side file for triage.
"""

from __future__ import annotations

import csv
import gzip
import json
import logging
import os
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import Any, Iterable

import requests

# ============================================================================
# Configuration — edit before running.
# ============================================================================

CONFIG: dict[str, Any] = {
    "base_url":      "https://ingestion.peak.ai",     # spoke tenants: https://ingestion.<cluster-identifier>.peak.ai
    "auth_token":    os.environ["PEAK_AUTH_TOKEN"],   # export PEAK_AUTH_TOKEN=...
    "solution_name": "your-solution-name",

    # csv_path    — required. CSV file OR folder of CSV part-files (e.g.
    #               "./orders/" containing part-0001.csv, part-0002.csv, ...).
    # object_name — required. Fully qualified table name as rolled out —
    #               include any prefix/suffix from your rollout (for example
    #               "acme_customers_v1", not just "customers").
    # file_glob   — optional, default "*.csv". Only used when csv_path is a
    #               folder. Set to "*.csv.gz" for gzipped Snowflake unloads, or
    #               "*" for Redshift's extensionless part-files.
    "csv_path":    "./customers.csv",
    "object_name": "acme_customers_v1",
    "file_glob":   "*.csv",

    "operation_type": "APPEND",                       # APPEND is fastest for historical

    # Set False if operation_type is UPSERT and the SAME primary key can appear
    # in more than one row of your source. Concurrent workers finish in an
    # unpredictable order, so for duplicated keys you can't tell which version
    # ends up in the warehouse. See the note below.
    "parallel": True,
}

# Tuning — fixed for predictable behaviour against the API's limits.
MAX_ROWS_PER_BATCH      = 2000
MAX_BYTES_PER_BATCH     = 1_048_576                   # 1 MB
RATE_LIMIT_RPS          = 30                          # well under the 50 RPS cap
WORKERS                 = 6                           # threads used only when parallel=True
MAX_RETRIES             = 5
INITIAL_BACKOFF_SECONDS = 1.0
REQUEST_TIMEOUT_SECONDS = 60
CHECKPOINT_DIR          = "./checkpoints"
FAILED_ROWS_DIR         = "./failed_rows"

# ============================================================================
# NOTE — parallelism and operation type
# ----------------------------------------------------------------------------
# Keep parallel=True for speed if:
#   - operation_type is APPEND, OR
#   - operation_type is UPSERT and every primary key appears in only one row.
#
# Set parallel=False if operation_type is UPSERT and the SAME primary key can
# appear in more than one row of your source. Concurrent workers finish in
# unpredictable order, so you can't tell which version of a duplicated key
# will end up in the warehouse — the same load run twice can produce different
# results. Sending one batch at a time fixes that.
#
# Alternatively, dedupe the source before running (keep only the latest record
# per key) and leave parallel=True.
# ============================================================================

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("bulk_ingest")


# ----------------------------------------------------------------------------
# Token bucket — thread-safe, shared across all workers.
# ----------------------------------------------------------------------------
class RateLimiter:
    def __init__(self, rps: int) -> None:
        self.rps = rps
        self.tokens = float(rps)
        self.last_refill = time.monotonic()
        self.lock = threading.Lock()

    def acquire(self) -> None:
        while True:
            with self.lock:
                now = time.monotonic()
                self.tokens = min(self.rps, self.tokens + (now - self.last_refill) * self.rps)
                self.last_refill = now
                if self.tokens >= 1:
                    self.tokens -= 1
                    return
                wait = (1 - self.tokens) / self.rps
            time.sleep(wait)


# ----------------------------------------------------------------------------
# Schema fetcher — auto-discovers column types for casting CSV strings.
# ----------------------------------------------------------------------------
def fetch_schema_types(base_url: str, auth_token: str, solution_name: str) -> dict[str, dict[str, str]]:
    url = f"{base_url}/api/v2/schema"
    headers = {"Authorization": auth_token}
    resp = requests.get(url, params={"solutionName": solution_name}, headers=headers, timeout=30)
    resp.raise_for_status()
    schema = resp.json()
    type_map: dict[str, dict[str, str]] = {}
    for table in schema.get("schema", []):
        name = table["objectName"].upper()
        type_map[name] = {
            col["columnName"]: col.get("dataType") or col.get("peakDataType") or "string"
            for col in table.get("columns", [])
        }
    return type_map


# ----------------------------------------------------------------------------
# Type casting — CSV string -> JSON value of the right type.
# ----------------------------------------------------------------------------
# Per-type casters. string/date/timestamp pass through unchanged (API validates format).
_CASTERS = {
    "integer": int,
    "numeric": float,
    "float":   float,
    "boolean": lambda s: s.lower() in ("true", "1", "t", "yes", "y"),
    "json":    json.loads,
}


def cast_value(value: str, data_type: str) -> Any:
    if value is None:
        return None
    # Treat standard warehouse null markers as None: empty string,
    # "\N" (Snowflake/Postgres default), and whitespace-only cells.
    stripped = value.strip()
    if stripped in ("", "\\N"):
        return None
    caster = _CASTERS.get((data_type or "string").lower())
    return caster(stripped) if caster else value


def cast_row(row: dict[str, str], column_types: dict[str, str]) -> dict[str, Any]:
    """Cast each cell to its schema type.

    If a cell cannot be cast (for example, "abc" in an integer column), the raw
    string is passed through unchanged. The API will then surface it as a
    structured failed-row error with the right code, rather than the worker
    thread crashing.
    """
    out: dict[str, Any] = {}
    for col, raw in row.items():
        try:
            out[col] = cast_value(raw, column_types.get(col, "string"))
        except (ValueError, TypeError, json.JSONDecodeError):
            out[col] = raw
    return out


# ----------------------------------------------------------------------------
# Source resolution — csv_path can be a single CSV file OR a folder of part-files
# (e.g. ./orders/ containing part-0001.csv, part-0002.csv). Folders are read in
# sorted filename order so the load is deterministic.
# ----------------------------------------------------------------------------
def resolve_csv_paths(csv_path: str, file_glob: str = "*.csv") -> list[str]:
    p = Path(csv_path)
    if p.is_dir():
        files = sorted(str(f) for f in p.glob(file_glob))
        if not files:
            raise FileNotFoundError(f"no files matching '{file_glob}' in folder: {csv_path}")
        return files
    return [csv_path]


# ----------------------------------------------------------------------------
# Streaming batch builder — reads every source file row-by-row and yields
# lists respecting both row count and size. Batches span file boundaries, so
# part-files don't produce undersized batches at their edges. Every file must
# carry its own header row.
# ----------------------------------------------------------------------------
def iter_batches(
    csv_paths: list[str],
    column_types: dict[str, str],
    max_rows: int,
    max_bytes: int,
) -> Iterable[list[dict[str, Any]]]:
    batch: list[dict[str, Any]] = []
    batch_bytes = 0
    for csv_path in csv_paths:
        opener = gzip.open if csv_path.endswith(".gz") else open
        with opener(csv_path, "rt", newline="", encoding="utf-8") as f:
            reader = csv.DictReader(f)
            for raw_row in reader:
                # DictReader puts cells beyond the header width under the None key.
                # That means the row is wider than the header — downstream values
                # would silently shift into the wrong column, so fail fast instead.
                if None in raw_row:
                    extras = raw_row[None]
                    raise RuntimeError(
                        f"{csv_path}: row has more columns than the header. "
                        f"Extra cells: {extras!r}. "
                        f"Check that every row matches the header width."
                    )
                row = cast_row(raw_row, column_types)
                row_bytes = len(json.dumps(row, default=str).encode("utf-8"))
                if batch and (len(batch) >= max_rows or batch_bytes + row_bytes >= max_bytes):
                    yield batch
                    batch, batch_bytes = [], 0
                batch.append(row)
                batch_bytes += row_bytes
    if batch:
        yield batch


# ----------------------------------------------------------------------------
# Per-batch ingest — retries 5xx and 429 with exponential backoff.
# ----------------------------------------------------------------------------
def ingest_batch(
    base_url: str,
    auth_token: str,
    object_name: str,
    solution_name: str,
    operation_type: str,
    batch: list[dict[str, Any]],
    *,
    rate_limiter: RateLimiter,
) -> dict[str, Any]:
    url = f"{base_url}/api/v2/objects/{object_name}"
    headers = {
        "Authorization": auth_token,
        "Content-Type": "application/json",
        "accept": "application/json",
    }
    body = {"solutionName": solution_name, "data": batch, "operationType": operation_type}

    backoff = INITIAL_BACKOFF_SECONDS
    for attempt in range(1, MAX_RETRIES + 1):
        rate_limiter.acquire()
        try:
            resp = requests.post(url, headers=headers, json=body, timeout=REQUEST_TIMEOUT_SECONDS)
        except requests.RequestException as exc:
            if attempt == MAX_RETRIES:
                raise
            log.warning("network error: %s; retry in %.1fs", exc, backoff)
            time.sleep(backoff)
            backoff *= 2
            continue

        # 200 = all rows ok; 207 = partial; 400 = all failed validation (terminal, do not retry)
        if resp.status_code in (200, 207, 400):
            return resp.json()

        # 429 / 5xx — back off and retry
        if resp.status_code == 429 or 500 <= resp.status_code < 600:
            if attempt == MAX_RETRIES:
                raise RuntimeError(f"max retries reached ({resp.status_code}): {resp.text[:200]}")
            log.warning("status %s; retry in %.1fs (attempt %d/%d)", resp.status_code, backoff, attempt, MAX_RETRIES)
            time.sleep(backoff)
            backoff *= 2
            continue

        # 401, 403, 404, 409 — terminal
        raise RuntimeError(f"{resp.status_code}: {resp.text[:200]}")

    raise RuntimeError("unreachable")


# ----------------------------------------------------------------------------
# Single-table runner — resume from checkpoint, parallelise batches.
# ----------------------------------------------------------------------------
def load_table(cfg: dict[str, Any], type_map: dict[str, dict[str, str]], rate_limiter: RateLimiter, workers: int) -> None:
    object_name = cfg["object_name"]
    file_glob = cfg.get("file_glob", "*.csv")
    csv_paths = resolve_csv_paths(cfg["csv_path"], file_glob)
    log.info("[%s] %d source file(s), glob=%s", object_name, len(csv_paths), file_glob)
    column_types = type_map.get(object_name.upper(), {})
    if not column_types:
        log.warning("[%s] no schema columns found; values pass through as strings", object_name)

    Path(CHECKPOINT_DIR).mkdir(parents=True, exist_ok=True)
    Path(FAILED_ROWS_DIR).mkdir(parents=True, exist_ok=True)
    ckpt_path = Path(CHECKPOINT_DIR) / f"{object_name}.json"
    failed_path = Path(FAILED_ROWS_DIR) / f"{object_name}.jsonl"

    next_batch = 0
    if ckpt_path.exists():
        next_batch = json.loads(ckpt_path.read_text()).get("next_batch", 0)
        if next_batch > 0:
            log.info("[%s] resuming from batch #%d", object_name, next_batch)

    # Skip already-completed batches by index; still must stream past them.
    pending = [
        (i, b) for i, b in enumerate(
            iter_batches(csv_paths, column_types, MAX_ROWS_PER_BATCH, MAX_BYTES_PER_BATCH)
        ) if i >= next_batch
    ]
    if not pending:
        log.info("[%s] already complete", object_name)
        return
    log.info("[%s] %d batches remaining", object_name, len(pending))

    total_ok = total_failed = 0
    ckpt_lock = threading.Lock()
    completed: set[int] = set()
    to_persist = next_batch

    def submit(idx_batch):
        idx, batch = idx_batch
        return idx, ingest_batch(
            cfg["base_url"], cfg["auth_token"], object_name, cfg["solution_name"],
            cfg["operation_type"], batch, rate_limiter=rate_limiter,
        )

    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(submit, ib): ib for ib in pending}
        for fut in as_completed(futures):
            idx, result = fut.result()
            ok = result.get("successRows", 0)
            failed = result.get("failedRows", 0)
            total_ok += ok
            total_failed += failed

            if result.get("failed"):
                with failed_path.open("a") as f:
                    for row in result["failed"]:
                        f.write(json.dumps({"batch": idx, **row}) + "\n")

            with ckpt_lock:
                completed.add(idx)
                while to_persist in completed:
                    completed.remove(to_persist)
                    to_persist += 1
                ckpt_path.write_text(json.dumps({"next_batch": to_persist}))

            log.info(
                "[%s] batch #%d: %d ok / %d failed (running: %d ok / %d failed)",
                object_name, idx, ok, failed, total_ok, total_failed,
            )

    log.info("[%s] complete: %d rows ingested, %d failed", object_name, total_ok, total_failed)


def main() -> None:
    cfg = CONFIG
    log.info("Fetching schema column types for solution %s", cfg["solution_name"])
    type_map = fetch_schema_types(cfg["base_url"], cfg["auth_token"], cfg["solution_name"])
    log.info("Schema fetched: %d tables", len(type_map))
    rate_limiter = RateLimiter(RATE_LIMIT_RPS)

    # parallel=False forces a single worker so batches are applied in submission
    # order — needed for UPSERT loads where a primary key can repeat across batches.
    workers = WORKERS if cfg.get("parallel", True) else 1
    log.info("Concurrency: %s", f"{workers} workers" if workers > 1 else "serial (batches applied in order)")

    log.info("=" * 70)
    log.info("Loading %s from %s", cfg["object_name"], cfg["csv_path"])
    log.info("=" * 70)
    load_table(cfg, type_map, rate_limiter, workers)
    log.info("Done.")


if __name__ == "__main__":
    main()
#!/usr/bin/env python3
"""
Bulk-ingest historical CSV data into the Peak Data Ingestion v2 API.

- One run loads one table from a CSV file or a folder of CSV part-files.
- Streams the source row-by-row (no full in-memory load).
- Dynamically sizes batches to <=2000 rows AND <=1 MB serialized.
- Concurrent workers share a token-bucket rate limiter.
- Exponential backoff on 5xx / 429; never retries 4xx.
- Checkpoint lets multi-hour runs resume cleanly after a crash.
- Failed rows from 207 responses written to a side file for triage.
"""

from __future__ import annotations

import csv
import gzip
import json
import logging
import os
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import Any, Iterable

import requests

# ============================================================================
# Configuration — edit before running.
# ============================================================================

CONFIG: dict[str, Any] = {
    "base_url":      "https://ingestion.peak.ai",     # spoke tenants: https://ingestion.<cluster-identifier>.peak.ai
    "auth_token":    os.environ["PEAK_AUTH_TOKEN"],   # export PEAK_AUTH_TOKEN=...
    "solution_name": "your-solution-name",

    # csv_path    — required. CSV file OR folder of CSV part-files (e.g.
    #               "./orders/" containing part-0001.csv, part-0002.csv, ...).
    # object_name — required. Fully qualified table name as rolled out —
    #               include any prefix/suffix from your rollout (for example
    #               "acme_customers_v1", not just "customers").
    # file_glob   — optional, default "*.csv". Only used when csv_path is a
    #               folder. Set to "*.csv.gz" for gzipped Snowflake unloads, or
    #               "*" for Redshift's extensionless part-files.
    "csv_path":    "./customers.csv",
    "object_name": "acme_customers_v1",
    "file_glob":   "*.csv",

    "operation_type": "APPEND",                       # APPEND is fastest for historical

    # Set False if operation_type is UPSERT and the SAME primary key can appear
    # in more than one row of your source. Concurrent workers finish in an
    # unpredictable order, so for duplicated keys you can't tell which version
    # ends up in the warehouse. See the note below.
    "parallel": True,
}

# Tuning — fixed for predictable behaviour against the API's limits.
MAX_ROWS_PER_BATCH      = 2000
MAX_BYTES_PER_BATCH     = 1_048_576                   # 1 MB
RATE_LIMIT_RPS          = 30                          # well under the 50 RPS cap
WORKERS                 = 6                           # threads used only when parallel=True
MAX_RETRIES             = 5
INITIAL_BACKOFF_SECONDS = 1.0
REQUEST_TIMEOUT_SECONDS = 60
CHECKPOINT_DIR          = "./checkpoints"
FAILED_ROWS_DIR         = "./failed_rows"

# ============================================================================
# NOTE — parallelism and operation type
# ----------------------------------------------------------------------------
# Keep parallel=True for speed if:
#   - operation_type is APPEND, OR
#   - operation_type is UPSERT and every primary key appears in only one row.
#
# Set parallel=False if operation_type is UPSERT and the SAME primary key can
# appear in more than one row of your source. Concurrent workers finish in
# unpredictable order, so you can't tell which version of a duplicated key
# will end up in the warehouse — the same load run twice can produce different
# results. Sending one batch at a time fixes that.
#
# Alternatively, dedupe the source before running (keep only the latest record
# per key) and leave parallel=True.
# ============================================================================

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("bulk_ingest")


# ----------------------------------------------------------------------------
# Token bucket — thread-safe, shared across all workers.
# ----------------------------------------------------------------------------
class RateLimiter:
    def __init__(self, rps: int) -> None:
        self.rps = rps
        self.tokens = float(rps)
        self.last_refill = time.monotonic()
        self.lock = threading.Lock()

    def acquire(self) -> None:
        while True:
            with self.lock:
                now = time.monotonic()
                self.tokens = min(self.rps, self.tokens + (now - self.last_refill) * self.rps)
                self.last_refill = now
                if self.tokens >= 1:
                    self.tokens -= 1
                    return
                wait = (1 - self.tokens) / self.rps
            time.sleep(wait)


# ----------------------------------------------------------------------------
# Schema fetcher — auto-discovers column types for casting CSV strings.
# ----------------------------------------------------------------------------
def fetch_schema_types(base_url: str, auth_token: str, solution_name: str) -> dict[str, dict[str, str]]:
    url = f"{base_url}/api/v2/schema"
    headers = {"Authorization": auth_token}
    resp = requests.get(url, params={"solutionName": solution_name}, headers=headers, timeout=30)
    resp.raise_for_status()
    schema = resp.json()
    type_map: dict[str, dict[str, str]] = {}
    for table in schema.get("schema", []):
        name = table["objectName"].upper()
        type_map[name] = {
            col["columnName"]: col.get("dataType") or col.get("peakDataType") or "string"
            for col in table.get("columns", [])
        }
    return type_map


# ----------------------------------------------------------------------------
# Type casting — CSV string -> JSON value of the right type.
# ----------------------------------------------------------------------------
# Per-type casters. string/date/timestamp pass through unchanged (API validates format).
_CASTERS = {
    "integer": int,
    "numeric": float,
    "float":   float,
    "boolean": lambda s: s.lower() in ("true", "1", "t", "yes", "y"),
    "json":    json.loads,
}


def cast_value(value: str, data_type: str) -> Any:
    if value is None:
        return None
    # Treat standard warehouse null markers as None: empty string,
    # "\N" (Snowflake/Postgres default), and whitespace-only cells.
    stripped = value.strip()
    if stripped in ("", "\\N"):
        return None
    caster = _CASTERS.get((data_type or "string").lower())
    return caster(stripped) if caster else value


def cast_row(row: dict[str, str], column_types: dict[str, str]) -> dict[str, Any]:
    """Cast each cell to its schema type.

    If a cell cannot be cast (for example, "abc" in an integer column), the raw
    string is passed through unchanged. The API will then surface it as a
    structured failed-row error with the right code, rather than the worker
    thread crashing.
    """
    out: dict[str, Any] = {}
    for col, raw in row.items():
        try:
            out[col] = cast_value(raw, column_types.get(col, "string"))
        except (ValueError, TypeError, json.JSONDecodeError):
            out[col] = raw
    return out


# ----------------------------------------------------------------------------
# Source resolution — csv_path can be a single CSV file OR a folder of part-files
# (e.g. ./orders/ containing part-0001.csv, part-0002.csv). Folders are read in
# sorted filename order so the load is deterministic.
# ----------------------------------------------------------------------------
def resolve_csv_paths(csv_path: str, file_glob: str = "*.csv") -> list[str]:
    p = Path(csv_path)
    if p.is_dir():
        files = sorted(str(f) for f in p.glob(file_glob))
        if not files:
            raise FileNotFoundError(f"no files matching '{file_glob}' in folder: {csv_path}")
        return files
    return [csv_path]


# ----------------------------------------------------------------------------
# Streaming batch builder — reads every source file row-by-row and yields
# lists respecting both row count and size. Batches span file boundaries, so
# part-files don't produce undersized batches at their edges. Every file must
# carry its own header row.
# ----------------------------------------------------------------------------
def iter_batches(
    csv_paths: list[str],
    column_types: dict[str, str],
    max_rows: int,
    max_bytes: int,
) -> Iterable[list[dict[str, Any]]]:
    batch: list[dict[str, Any]] = []
    batch_bytes = 0
    for csv_path in csv_paths:
        opener = gzip.open if csv_path.endswith(".gz") else open
        with opener(csv_path, "rt", newline="", encoding="utf-8") as f:
            reader = csv.DictReader(f)
            for raw_row in reader:
                # DictReader puts cells beyond the header width under the None key.
                # That means the row is wider than the header — downstream values
                # would silently shift into the wrong column, so fail fast instead.
                if None in raw_row:
                    extras = raw_row[None]
                    raise RuntimeError(
                        f"{csv_path}: row has more columns than the header. "
                        f"Extra cells: {extras!r}. "
                        f"Check that every row matches the header width."
                    )
                row = cast_row(raw_row, column_types)
                row_bytes = len(json.dumps(row, default=str).encode("utf-8"))
                if batch and (len(batch) >= max_rows or batch_bytes + row_bytes >= max_bytes):
                    yield batch
                    batch, batch_bytes = [], 0
                batch.append(row)
                batch_bytes += row_bytes
    if batch:
        yield batch


# ----------------------------------------------------------------------------
# Per-batch ingest — retries 5xx and 429 with exponential backoff.
# ----------------------------------------------------------------------------
def ingest_batch(
    base_url: str,
    auth_token: str,
    object_name: str,
    solution_name: str,
    operation_type: str,
    batch: list[dict[str, Any]],
    *,
    rate_limiter: RateLimiter,
) -> dict[str, Any]:
    url = f"{base_url}/api/v2/objects/{object_name}"
    headers = {
        "Authorization": auth_token,
        "Content-Type": "application/json",
        "accept": "application/json",
    }
    body = {"solutionName": solution_name, "data": batch, "operationType": operation_type}

    backoff = INITIAL_BACKOFF_SECONDS
    for attempt in range(1, MAX_RETRIES + 1):
        rate_limiter.acquire()
        try:
            resp = requests.post(url, headers=headers, json=body, timeout=REQUEST_TIMEOUT_SECONDS)
        except requests.RequestException as exc:
            if attempt == MAX_RETRIES:
                raise
            log.warning("network error: %s; retry in %.1fs", exc, backoff)
            time.sleep(backoff)
            backoff *= 2
            continue

        # 200 = all rows ok; 207 = partial; 400 = all failed validation (terminal, do not retry)
        if resp.status_code in (200, 207, 400):
            return resp.json()

        # 429 / 5xx — back off and retry
        if resp.status_code == 429 or 500 <= resp.status_code < 600:
            if attempt == MAX_RETRIES:
                raise RuntimeError(f"max retries reached ({resp.status_code}): {resp.text[:200]}")
            log.warning("status %s; retry in %.1fs (attempt %d/%d)", resp.status_code, backoff, attempt, MAX_RETRIES)
            time.sleep(backoff)
            backoff *= 2
            continue

        # 401, 403, 404, 409 — terminal
        raise RuntimeError(f"{resp.status_code}: {resp.text[:200]}")

    raise RuntimeError("unreachable")


# ----------------------------------------------------------------------------
# Single-table runner — resume from checkpoint, parallelise batches.
# ----------------------------------------------------------------------------
def load_table(cfg: dict[str, Any], type_map: dict[str, dict[str, str]], rate_limiter: RateLimiter, workers: int) -> None:
    object_name = cfg["object_name"]
    file_glob = cfg.get("file_glob", "*.csv")
    csv_paths = resolve_csv_paths(cfg["csv_path"], file_glob)
    log.info("[%s] %d source file(s), glob=%s", object_name, len(csv_paths), file_glob)
    column_types = type_map.get(object_name.upper(), {})
    if not column_types:
        log.warning("[%s] no schema columns found; values pass through as strings", object_name)

    Path(CHECKPOINT_DIR).mkdir(parents=True, exist_ok=True)
    Path(FAILED_ROWS_DIR).mkdir(parents=True, exist_ok=True)
    ckpt_path = Path(CHECKPOINT_DIR) / f"{object_name}.json"
    failed_path = Path(FAILED_ROWS_DIR) / f"{object_name}.jsonl"

    next_batch = 0
    if ckpt_path.exists():
        next_batch = json.loads(ckpt_path.read_text()).get("next_batch", 0)
        if next_batch > 0:
            log.info("[%s] resuming from batch #%d", object_name, next_batch)

    # Skip already-completed batches by index; still must stream past them.
    pending = [
        (i, b) for i, b in enumerate(
            iter_batches(csv_paths, column_types, MAX_ROWS_PER_BATCH, MAX_BYTES_PER_BATCH)
        ) if i >= next_batch
    ]
    if not pending:
        log.info("[%s] already complete", object_name)
        return
    log.info("[%s] %d batches remaining", object_name, len(pending))

    total_ok = total_failed = 0
    ckpt_lock = threading.Lock()
    completed: set[int] = set()
    to_persist = next_batch

    def submit(idx_batch):
        idx, batch = idx_batch
        return idx, ingest_batch(
            cfg["base_url"], cfg["auth_token"], object_name, cfg["solution_name"],
            cfg["operation_type"], batch, rate_limiter=rate_limiter,
        )

    with ThreadPoolExecutor(max_workers=workers) as pool:
        futures = {pool.submit(submit, ib): ib for ib in pending}
        for fut in as_completed(futures):
            idx, result = fut.result()
            ok = result.get("successRows", 0)
            failed = result.get("failedRows", 0)
            total_ok += ok
            total_failed += failed

            if result.get("failed"):
                with failed_path.open("a") as f:
                    for row in result["failed"]:
                        f.write(json.dumps({"batch": idx, **row}) + "\n")

            with ckpt_lock:
                completed.add(idx)
                while to_persist in completed:
                    completed.remove(to_persist)
                    to_persist += 1
                ckpt_path.write_text(json.dumps({"next_batch": to_persist}))

            log.info(
                "[%s] batch #%d: %d ok / %d failed (running: %d ok / %d failed)",
                object_name, idx, ok, failed, total_ok, total_failed,
            )

    log.info("[%s] complete: %d rows ingested, %d failed", object_name, total_ok, total_failed)


def main() -> None:
    cfg = CONFIG
    log.info("Fetching schema column types for solution %s", cfg["solution_name"])
    type_map = fetch_schema_types(cfg["base_url"], cfg["auth_token"], cfg["solution_name"])
    log.info("Schema fetched: %d tables", len(type_map))
    rate_limiter = RateLimiter(RATE_LIMIT_RPS)

    # parallel=False forces a single worker so batches are applied in submission
    # order — needed for UPSERT loads where a primary key can repeat across batches.
    workers = WORKERS if cfg.get("parallel", True) else 1
    log.info("Concurrency: %s", f"{workers} workers" if workers > 1 else "serial (batches applied in order)")

    log.info("=" * 70)
    log.info("Loading %s from %s", cfg["object_name"], cfg["csv_path"])
    log.info("=" * 70)
    load_table(cfg, type_map, rate_limiter, workers)
    log.info("Done.")


if __name__ == "__main__":
    main()