UiPath Documentation
industry-department-solutions
latest
false
Supply Chain & Retail Solutions API guide
  • Overview
    • Introduction
    • Getting Started
    • Integration walkthrough
    • API Guide
    • Schema lifecycle
    • Object-level validations
    • Scheduled ingestion
    • Historical data ingestion
    • Data Quality Dashboard
    • Customizations
    • Data Onboarding Checklist
  • API Resources

Historical data ingestion

Load large volumes of historical data into the v2 Supply Chain & Retail Solutions Data Ingestion API.

This page describes how to load large volumes of historical data (millions to billions of rows) into the v2 Data Ingestion API. It provides a reference Python script that loads every table in your solution in a single run and handles streaming, batching, rate limiting, retries, and resume — along with the warehouse-specific configuration you'll need to give it.

Scope. This script is for one-shot historical loads — a single, possibly multi-hour load performed once during onboarding. For ongoing incremental data after that initial load, post new rows directly to POST /api/v2/objects/{objectName} using the contracts in the API Guide. The script's checkpoint is a crash-recovery aid within a single load, not a state machine across separate runs over time.

For everyday low-volume ingestion, the API contracts described in Getting Started and the API Guide are sufficient. Use this page when you have a one-time bulk load to perform.

Before you start​

  • Your solution's tables have been rolled out (see Schema lifecycle).
  • You have a Personal Access Token (see Getting Started).
  • Each table you're loading is available as CSV — either a single file, or a folder of CSV part-files (as produced by chunked warehouse exports).
  • Every CSV file includes a header row, and the column names match the rolled-out schema (case-insensitive). When the source is a folder, every file in it must share the same header.
  • You know whether your tenant is on the hub (https://ingestion.peak.ai) or on a spoke cluster (https://ingestion.<cluster-identifier>.peak.ai — for example https://ingestion.peaksydney.peak.ai). Your account manager provides the cluster identifier for your tenant. See the warning below.
Warning:

Spoke tenants must set base_url to their spoke host before running the script. The script defaults base_url to the hub host https://ingestion.peak.ai. A spoke tenant that leaves the default can violate the tenant's data-residency requirements.

Source file format​

The script reads CSV files. Headers are required — every CSV file must include a header row with the column names from the rolled-out schema (case-insensitive). The script matches columns by name, so the order of columns in your SELECT doesn't matter and there's no risk of accidentally writing quoted_price values into the discount_pct column. If csv_path is a folder of part-files, every file in it must share the same header.

Beyond headers, two things about your unload can vary, and the script handles both:

  • File naming pattern — set file_glob in CONFIG when csv_path is a folder (default *.csv).
  • Gzip compression — auto-detected from the .gz extension; no flag needed.

File or folder?​

csv_path accepts either form:

  • A file path — e.g. ./customers.csv. file_glob is ignored.
  • A folder path — e.g. ./customers/. Every file in the folder matching file_glob is loaded as if it were one logical CSV. Use this for chunked warehouse exports.

Folder-load file order. When csv_path is a folder, files are loaded in lexicographic (byte-by-byte ASCII) order of their filenames. This is the right order when the names already encode order as zero-padded numeric or ISO-style date prefixes — the form that warehouse exports produce by default.

FilenamesLexicographic orderRight?
part-0001.csv, part-0002.csv, …, part-9999.csvpart-0001, part-0002, …, part-9999✓ matches numeric order
2024-01-15.csv, 2024-02-10.csv, 2024-06-01.csv2024-01-15, 2024-02-10, 2024-06-01✓ matches chronological order
part-1.csv, part-2.csv, …, part-10.csvpart-1, part-10, part-2, …✗ 10 sorts before 2 (not zero-padded)
15-01-2023.csv, 10-02-2024.csv, 01-06-2024.csv (DD-MM-YYYY)01-06-2024, 10-02-2024, 15-01-2023✗ day-first puts June 2024 before Jan 2023
01-15-2023.csv, 02-10-2024.csv, 06-01-2024.csv (MM-DD-YYYY)01-15-2023, 02-10-2024, 06-01-2024✗ year is last, so 2023 and 2024 interleave
January_report.csv, February_report.csv, March_report.csvFebruary, January, March✗ alphabetical, not chronological

If file order doesn't matter (every file is an independent batch and rows have no cross-file dependency), this isn't a concern. If file order does matter — for example, you're using APPEND and the row sequence is meaningful, or UPSERT and later files supersede earlier ones — make sure your filenames sort lexicographically into the order you intend. The safe pattern is a zero-padded numeric or ISO-date prefix (e.g., part-0001.csv, 2024-06-01.csv); rename or re-export if your source uses something else.

Fully qualified object_name​

object_name must be the exact table name as rolled out — including any prefix or suffix that was applied at rollout. For example, if your rollout applied a prefix acme_ and suffix _v1, the configured object_name is acme_customers_v1, not customers. If your rollout supplied neither a prefix nor a suffix, object_name is just the table name itself (e.g., customers).

To look up the exact value for any table, call GET /api/v2/schema?solutionName=<your-solution> and read the objectName field for each entry — that's the exact string to use in object_name.

From Snowflake​

Snowflake COPY INTO @stage defaults to no headers, gzip compression, and \N as the null marker. Add HEADER = TRUE so columns match by name, and NULL_IF = ('') so nulls are written as empty strings instead of \N. (The script handles either marker, but empty strings give a cleaner CSV.)

COPY INTO @your_stage/orders/
FROM (SELECT custcode, addcode, postcode, streetcode, basketvalue FROM orders)
FILE_FORMAT = (TYPE = CSV HEADER = TRUE NULL_IF = (''));
COPY INTO @your_stage/orders/
FROM (SELECT custcode, addcode, postcode, streetcode, basketvalue FROM orders)
FILE_FORMAT = (TYPE = CSV HEADER = TRUE NULL_IF = (''));

Once the unloaded files are downloaded locally, add an entry to CONFIG["tables"] with csv_path set to that folder. Use "*.csv" for file_glob instead if you disabled gzip with COMPRESSION = NONE:

{"object_name": "acme_orders_v1", "csv_path": "./orders/", "file_glob": "*.csv.gz"},
{"object_name": "acme_orders_v1", "csv_path": "./orders/", "file_glob": "*.csv.gz"},

From Redshift​

Redshift UNLOAD defaults to no headers, no compression, and no file extension on the part-files. Add HEADER so columns match by name, and EXTENSION 'csv' so files are easy to glob:

UNLOAD ('SELECT custcode, addcode, postcode, streetcode, basketvalue FROM orders')
TO 's3://your-bucket/orders/'
HEADER
FORMAT CSV
EXTENSION 'csv';
UNLOAD ('SELECT custcode, addcode, postcode, streetcode, basketvalue FROM orders')
TO 's3://your-bucket/orders/'
HEADER
FORMAT CSV
EXTENSION 'csv';

Once the unloaded files are downloaded locally, add an entry to CONFIG["tables"] with csv_path set to that folder. "*.csv" is the default file_glob; use "*" if you kept Redshift's extensionless part-files, or "*.gz" if you enabled GZIP:

{"object_name": "acme_orders_v1", "csv_path": "./orders/", "file_glob": "*.csv"},
{"object_name": "acme_orders_v1", "csv_path": "./orders/", "file_glob": "*.csv"},

Null values and bad cells​

The script treats a cell as null when, after trimming whitespace, it matches one of these markers (case-insensitively): an empty string, \N (the Redshift UNLOAD and Snowflake COPY INTO defaults), NULL, NA, or N/A. This covers the null representations found in most real-world warehouse exports.

Warning:

NA and N/A are treated as null. That's convenient for most data, but wrong if a column can legitimately hold those strings — for example, NA is Namibia's ISO country code, and N/A is a valid category label in some datasets. If any column in your load can contain them as real values, edit the NULL_MARKERS set near the top of the script and remove the entries you don't want — whatever you remove is then sent through as that exact string.

The full set of null markers lives in one place so you can adjust it:

NULL_MARKERS = {"", "\\N", "NULL", "NA", "N/A"}
NULL_MARKERS = {"", "\\N", "NULL", "NA", "N/A"}

Uncastable cells are passed through as raw strings rather than crashing the load. If "abc" lands in an integer column, the API returns that single row as a failed row with the appropriate error code (see Validation behavior for the failed-row response shape) — one bad cell becomes one rejected row, not a stopped load.

Character encoding​

Customer exports aren't always UTF-8. Before reading each file the script detects its encoding, trying utf-8-sig (UTF-8, with any byte-order mark stripped), then cp1252 (common for Windows-generated files), then latin-1 as a catch-all that decodes any byte sequence.

When a file is decoded as anything other than utf-8-sig, the script logs which encoding it fell back to, so you can spot a mojibake risk early. No configuration is needed.

Run modes and options​

Behavior is controlled with command-line flags — you don't edit the CONFIG block to switch modes:

FlagEffect
--mode historicalDefault. APPEND operation, batches sent in parallel.
--mode dailyUPSERT operation, batches sent serially (one at a time, in order). For smaller recurring reloads.
--table OBJECT_NAMELoad only this one table from CONFIG["tables"] instead of all of them.
--dry-runParse, validate, and batch every table but send nothing to the API.
--archiveAfter each table loads successfully, move its source files into archive/<date>/<table>/.

--mode historical (APPEND, parallel)​

For a one-time bulk load, historical mode is the right choice — APPEND is significantly faster than UPSERT when you don't need update-by-primary-key semantics, and parallel workers maximise throughput.

One thing to note about APPEND: if a batch partially succeeds and you re-send the same batch, the rows that already landed will fail with a duplicate-primary-key error. The script's checkpointing handles this for you — a successful batch is never re-sent on a subsequent run.

--mode daily (UPSERT, serial)​

Daily mode uses UPSERT, for recurring incremental reloads where the same primary key can reappear and later rows should supersede earlier ones. Batches are sent serially — one at a time, in order — so the last write for a key wins deterministically. (Parallel workers finish in unpredictable order, so a parallel UPSERT of duplicated keys could leave either version in the warehouse; serial avoids that.)

--dry-run​

A dry run streams, parses, casts, and batches every table exactly as a real load would — surfacing malformed CSVs, too-wide rows, and encoding problems — but sends nothing and writes no checkpoint or failed-row files. It logs the row and batch count each table would produce.

Use it to validate a fresh set of exports before committing to a multi-hour load. (It still reads your schema, so it needs the token and network access; it makes no writes.)

Loading multiple tables​

The script loads every table in CONFIG["tables"] in a single run — you don't invoke it once per table. List one entry per table:

"tables": [
    {"object_name": "acme_products_v1",    "csv_path": "./products/"},
    {"object_name": "acme_orders_v1",       "csv_path": "./orders/",      "file_glob": "*.csv.gz"},
    {"object_name": "acme_order_items_v1",  "csv_path": "./order_items/"},
],
"tables": [
    {"object_name": "acme_products_v1",    "csv_path": "./products/"},
    {"object_name": "acme_orders_v1",       "csv_path": "./orders/",      "file_glob": "*.csv.gz"},
    {"object_name": "acme_order_items_v1",  "csv_path": "./order_items/"},
],

List order is load order — put parent tables before their children (foreign-key dependency order). In the example above, if order_items references orders and orders references products, listing them products → orders → order_items loads them in the right order.

Loading them out of order causes foreign-key violations in the asynchronous validation stage. Tables load sequentially (one finishes before the next starts); batches within a table run in parallel in historical mode.

To load a single table from the list — for a re-run, or to load them one at a time — pass --table <object_name>.

Reference script​

The script is self-contained — paste it into a single Python file, edit the CONFIG block at the top, and run it. It requires only the requests library (pip install requests).

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

- One run loads every table in CONFIG["tables"], in list order.
- Streams each 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 written to a JSONL log and a consolidated JSON report.
- Run mode, single-table, dry-run, and archive are chosen with CLI flags.
"""

from __future__ import annotations

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

import requests

CONFIG: dict[str, Any] = {
    "base_url":      "https://ingestion.peak.ai",
    "auth_token":    os.environ["PEAK_AUTH_TOKEN"],
    "solution_name": "your-solution-name",

    "tables": [
        {"object_name": "acme_customers_v1", "csv_path": "./customers.csv"},
        {"object_name": "acme_orders_v1",    "csv_path": "./orders/", "file_glob": "*.csv.gz"},
    ],
}

MAX_ROWS_PER_BATCH = 2000
MAX_BYTES_PER_BATCH = 1_048_576
RATE_LIMIT_RPS = 30
WORKERS = 6
MAX_RETRIES = 5
INITIAL_BACKOFF_SECONDS = 1.0
REQUEST_TIMEOUT_SECONDS = 60
CHECKPOINT_DIR = "./checkpoints"
FAILED_ROWS_DIR = "./failed_rows"
ARCHIVE_DIR = "./archive"

MODES = {
    "historical": {"operation_type": "APPEND", "parallel": True},
    "daily":      {"operation_type": "UPSERT", "parallel": False},
}

NULL_MARKERS = {"", "\\N", "NULL", "NA", "N/A"}
_NULL_MARKERS_LOWER = {m.lower() for m in NULL_MARKERS}

ENCODINGS = ("utf-8-sig", "cp1252", "latin-1")

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

@dataclass
class Run:
    """Everything a single load run needs beyond the per-table config."""
    base_url: str
    auth_token: str
    solution_name: str
    operation_type: str
    workers: int
    dry_run: bool
    archive: bool
    rate_limiter: "RateLimiter"

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)

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

_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
    stripped = value.strip()
    if stripped.lower() in _NULL_MARKERS_LOWER:
        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

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]

def detect_encoding(csv_path: str) -> str:
    opener = gzip.open if csv_path.endswith(".gz") else open
    for enc in ENCODINGS:
        try:
            with opener(csv_path, "rt", encoding=enc, newline="") as f:
                for _ in f:
                    pass
            return enc
        except UnicodeDecodeError:
            continue
    return ENCODINGS[-1]

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
        encoding = detect_encoding(csv_path)
        if encoding != ENCODINGS[0]:
            log.info("%s: decoded as %s (not %s)", csv_path, encoding, ENCODINGS[0])
        with opener(csv_path, "rt", newline="", encoding=encoding) as f:
            reader = csv.DictReader(f)
            for raw_row in reader:
                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

def ingest_batch(run: Run, object_name: str, batch: list[dict[str, Any]]) -> dict[str, Any]:
    url = f"{run.base_url}/api/v2/objects/{object_name}"
    headers = {
        "Authorization": run.auth_token,
        "Content-Type": "application/json",
        "accept": "application/json",
    }
    body = {"solutionName": run.solution_name, "data": batch, "operationType": run.operation_type}

    backoff = INITIAL_BACKOFF_SECONDS
    for attempt in range(1, MAX_RETRIES + 1):
        run.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

        if resp.status_code in (200, 207, 400):
            return resp.json()

        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

        raise RuntimeError(f"{resp.status_code}: {resp.text[:200]}")

    raise RuntimeError("unreachable")

def write_failed_report(failed_jsonl: Path, report_path: Path) -> int:
    rows = [json.loads(line) for line in failed_jsonl.read_text().splitlines() if line.strip()]
    report_path.write_text(json.dumps(rows, indent=2, default=str))
    return len(rows)

def archive_source(csv_paths: list[str], object_name: str) -> Path:
    stamp = datetime.now().strftime("%Y-%m-%d")
    dest = Path(ARCHIVE_DIR) / stamp / object_name
    dest.mkdir(parents=True, exist_ok=True)
    for path in csv_paths:
        shutil.move(path, str(dest / Path(path).name))
    return dest

def load_table(run: Run, table: dict[str, Any], type_map: dict[str, dict[str, str]]) -> None:
    object_name = table["object_name"]
    file_glob = table.get("file_glob", "*.csv")
    csv_paths = resolve_csv_paths(table["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)

    if run.dry_run:
        rows = batches = 0
        for batch in iter_batches(csv_paths, column_types, MAX_ROWS_PER_BATCH, MAX_BYTES_PER_BATCH):
            batches += 1
            rows += len(batch)
        log.info("[%s] DRY RUN: %d rows would be sent in %d batches — nothing written", object_name, rows, batches)
        return

    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)

    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)
        if run.archive:
            log.info("[%s] archived to %s", object_name, archive_source(csv_paths, 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(run, object_name, batch)

    with ThreadPoolExecutor(max_workers=run.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,
            )

    if failed_path.exists():
        report_path = Path(FAILED_ROWS_DIR) / f"{object_name}_failed.json"
        count = write_failed_report(failed_path, report_path)
        log.info("[%s] %d failed row(s) written to %s", object_name, count, report_path)

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

    if run.archive:
        log.info("[%s] archived to %s", object_name, archive_source(csv_paths, object_name))

def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Bulk-load historical CSV data into the Peak Data Ingestion v2 API.")
    parser.add_argument(
        "--mode", choices=sorted(MODES), default="historical",
        help="historical = APPEND + parallel (default); daily = UPSERT + serial.",
    )
    parser.add_argument(
        "--table", metavar="OBJECT_NAME",
        help="Load only this table from CONFIG['tables']. Default: load them all, in order.",
    )
    parser.add_argument(
        "--dry-run", action="store_true",
        help="Parse, validate and batch every table but send nothing to the API.",
    )
    parser.add_argument(
        "--archive", action="store_true",
        help="After a table loads successfully, move its source files into archive/<date>/<table>/.",
    )
    return parser.parse_args()

def main() -> None:
    args = parse_args()
    cfg = CONFIG
    mode = MODES[args.mode]

    tables = cfg["tables"]
    if args.table:
        tables = [t for t in tables if t["object_name"] == args.table]
        if not tables:
            names = ", ".join(t["object_name"] for t in cfg["tables"])
            raise SystemExit(f"--table {args.table!r} not found in CONFIG['tables'] (have: {names})")

    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))

    workers = WORKERS if mode["parallel"] else 1
    run = Run(
        base_url=cfg["base_url"],
        auth_token=cfg["auth_token"],
        solution_name=cfg["solution_name"],
        operation_type=mode["operation_type"],
        workers=workers,
        dry_run=args.dry_run,
        archive=args.archive,
        rate_limiter=RateLimiter(RATE_LIMIT_RPS),
    )
    log.info(
        "Mode: %s (%s, %s)%s",
        args.mode,
        run.operation_type,
        f"{workers} workers" if workers > 1 else "serial — batches applied in order",
        " [DRY RUN]" if run.dry_run else "",
    )

    for table in tables:
        log.info("=" * 70)
        log.info("Loading %s from %s", table["object_name"], table["csv_path"])
        log.info("=" * 70)
        load_table(run, table, type_map)
    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 every table in CONFIG["tables"], in list order.
- Streams each 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 written to a JSONL log and a consolidated JSON report.
- Run mode, single-table, dry-run, and archive are chosen with CLI flags.
"""

from __future__ import annotations

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

import requests

CONFIG: dict[str, Any] = {
    "base_url":      "https://ingestion.peak.ai",
    "auth_token":    os.environ["PEAK_AUTH_TOKEN"],
    "solution_name": "your-solution-name",

    "tables": [
        {"object_name": "acme_customers_v1", "csv_path": "./customers.csv"},
        {"object_name": "acme_orders_v1",    "csv_path": "./orders/", "file_glob": "*.csv.gz"},
    ],
}

MAX_ROWS_PER_BATCH = 2000
MAX_BYTES_PER_BATCH = 1_048_576
RATE_LIMIT_RPS = 30
WORKERS = 6
MAX_RETRIES = 5
INITIAL_BACKOFF_SECONDS = 1.0
REQUEST_TIMEOUT_SECONDS = 60
CHECKPOINT_DIR = "./checkpoints"
FAILED_ROWS_DIR = "./failed_rows"
ARCHIVE_DIR = "./archive"

MODES = {
    "historical": {"operation_type": "APPEND", "parallel": True},
    "daily":      {"operation_type": "UPSERT", "parallel": False},
}

NULL_MARKERS = {"", "\\N", "NULL", "NA", "N/A"}
_NULL_MARKERS_LOWER = {m.lower() for m in NULL_MARKERS}

ENCODINGS = ("utf-8-sig", "cp1252", "latin-1")

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

@dataclass
class Run:
    """Everything a single load run needs beyond the per-table config."""
    base_url: str
    auth_token: str
    solution_name: str
    operation_type: str
    workers: int
    dry_run: bool
    archive: bool
    rate_limiter: "RateLimiter"

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)

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

_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
    stripped = value.strip()
    if stripped.lower() in _NULL_MARKERS_LOWER:
        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

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]

def detect_encoding(csv_path: str) -> str:
    opener = gzip.open if csv_path.endswith(".gz") else open
    for enc in ENCODINGS:
        try:
            with opener(csv_path, "rt", encoding=enc, newline="") as f:
                for _ in f:
                    pass
            return enc
        except UnicodeDecodeError:
            continue
    return ENCODINGS[-1]

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
        encoding = detect_encoding(csv_path)
        if encoding != ENCODINGS[0]:
            log.info("%s: decoded as %s (not %s)", csv_path, encoding, ENCODINGS[0])
        with opener(csv_path, "rt", newline="", encoding=encoding) as f:
            reader = csv.DictReader(f)
            for raw_row in reader:
                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

def ingest_batch(run: Run, object_name: str, batch: list[dict[str, Any]]) -> dict[str, Any]:
    url = f"{run.base_url}/api/v2/objects/{object_name}"
    headers = {
        "Authorization": run.auth_token,
        "Content-Type": "application/json",
        "accept": "application/json",
    }
    body = {"solutionName": run.solution_name, "data": batch, "operationType": run.operation_type}

    backoff = INITIAL_BACKOFF_SECONDS
    for attempt in range(1, MAX_RETRIES + 1):
        run.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

        if resp.status_code in (200, 207, 400):
            return resp.json()

        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

        raise RuntimeError(f"{resp.status_code}: {resp.text[:200]}")

    raise RuntimeError("unreachable")

def write_failed_report(failed_jsonl: Path, report_path: Path) -> int:
    rows = [json.loads(line) for line in failed_jsonl.read_text().splitlines() if line.strip()]
    report_path.write_text(json.dumps(rows, indent=2, default=str))
    return len(rows)

def archive_source(csv_paths: list[str], object_name: str) -> Path:
    stamp = datetime.now().strftime("%Y-%m-%d")
    dest = Path(ARCHIVE_DIR) / stamp / object_name
    dest.mkdir(parents=True, exist_ok=True)
    for path in csv_paths:
        shutil.move(path, str(dest / Path(path).name))
    return dest

def load_table(run: Run, table: dict[str, Any], type_map: dict[str, dict[str, str]]) -> None:
    object_name = table["object_name"]
    file_glob = table.get("file_glob", "*.csv")
    csv_paths = resolve_csv_paths(table["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)

    if run.dry_run:
        rows = batches = 0
        for batch in iter_batches(csv_paths, column_types, MAX_ROWS_PER_BATCH, MAX_BYTES_PER_BATCH):
            batches += 1
            rows += len(batch)
        log.info("[%s] DRY RUN: %d rows would be sent in %d batches — nothing written", object_name, rows, batches)
        return

    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)

    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)
        if run.archive:
            log.info("[%s] archived to %s", object_name, archive_source(csv_paths, 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(run, object_name, batch)

    with ThreadPoolExecutor(max_workers=run.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,
            )

    if failed_path.exists():
        report_path = Path(FAILED_ROWS_DIR) / f"{object_name}_failed.json"
        count = write_failed_report(failed_path, report_path)
        log.info("[%s] %d failed row(s) written to %s", object_name, count, report_path)

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

    if run.archive:
        log.info("[%s] archived to %s", object_name, archive_source(csv_paths, object_name))

def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Bulk-load historical CSV data into the Peak Data Ingestion v2 API.")
    parser.add_argument(
        "--mode", choices=sorted(MODES), default="historical",
        help="historical = APPEND + parallel (default); daily = UPSERT + serial.",
    )
    parser.add_argument(
        "--table", metavar="OBJECT_NAME",
        help="Load only this table from CONFIG['tables']. Default: load them all, in order.",
    )
    parser.add_argument(
        "--dry-run", action="store_true",
        help="Parse, validate and batch every table but send nothing to the API.",
    )
    parser.add_argument(
        "--archive", action="store_true",
        help="After a table loads successfully, move its source files into archive/<date>/<table>/.",
    )
    return parser.parse_args()

def main() -> None:
    args = parse_args()
    cfg = CONFIG
    mode = MODES[args.mode]

    tables = cfg["tables"]
    if args.table:
        tables = [t for t in tables if t["object_name"] == args.table]
        if not tables:
            names = ", ".join(t["object_name"] for t in cfg["tables"])
            raise SystemExit(f"--table {args.table!r} not found in CONFIG['tables'] (have: {names})")

    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))

    workers = WORKERS if mode["parallel"] else 1
    run = Run(
        base_url=cfg["base_url"],
        auth_token=cfg["auth_token"],
        solution_name=cfg["solution_name"],
        operation_type=mode["operation_type"],
        workers=workers,
        dry_run=args.dry_run,
        archive=args.archive,
        rate_limiter=RateLimiter(RATE_LIMIT_RPS),
    )
    log.info(
        "Mode: %s (%s, %s)%s",
        args.mode,
        run.operation_type,
        f"{workers} workers" if workers > 1 else "serial — batches applied in order",
        " [DRY RUN]" if run.dry_run else "",
    )

    for table in tables:
        log.info("=" * 70)
        log.info("Loading %s from %s", table["object_name"], table["csv_path"])
        log.info("=" * 70)
        load_table(run, table, type_map)
    log.info("Done.")

if __name__ == "__main__":
    main()

Running the script​

  1. Save the script as bulk_ingest.py.
  2. Install the dependency: pip install requests.
  3. Export your Personal Access Token: export PEAK_AUTH_TOKEN=....
  4. Edit the CONFIG block:
    • Set solution_name and the tables list (one entry per table, in foreign-key dependency order).
    • Set file_glob on an entry if that table's source is a folder with non-default file names.
    • If your tenant is on a spoke cluster, set base_url to your spoke host (https://ingestion.<cluster-identifier>.peak.ai) — see the spoke base-URL warning above.
  5. Validate first with a dry run: python bulk_ingest.py --dry-run.
  6. Run it: python bulk_ingest.py (add --mode daily, --table <name>, or --archive as needed — see Run modes and options).

Result​

The script logs progress per batch, table by table. You can stop it (Ctrl-C) and resume it later — it picks up from the last successfully-persisted batch index.

Rows the API rejects during the load are captured two ways, per table, in the failed_rows/ directory:

  • failed_rows/<table>.jsonl — the durable, crash-safe log, appended one line per rejected row as the load runs.
  • failed_rows/<table>_failed.json — a consolidated JSON array written when the table finishes, for quickly eyeballing everything that failed and why.

Asynchronous validation failures (foreign-key violations, primary-key collisions against pre-existing data) don't appear in either file — they show up in the Data Quality Dashboard instead.

With --archive, each table's source files are moved into archive/<date>/<table>/ once it loads successfully, leaving you an empty source folder to drop the next batch of files into and a dated record of what was already ingested. (Re-running an archived table reports "no files matching" — that's expected; its files have moved.)

Resuming after a crash​

If the script stops mid-load — a network blip, a laptop reboot, anything — just run it again and it picks up from where it left off. No rows are re-sent and none are dropped.

Two files persist between runs to make that work:

  • checkpoints/<table>.json — tracks how far the load got, so the script knows where to resume.
  • failed_rows/<table>.jsonl — appended to each time the API rejects a row.

On resume, the script reads the checkpoint, skips the batches it already completed, and continues from there. You'll see [<table>] resuming from batch #N in the logs at startup.

If you want a clean restart — for example, you've truncated the warehouse table and want to load from scratch — delete the state files first. Replace <table> with the object_name from your CONFIG (e.g., acme_customers_v1):

rm checkpoints/<table>.json failed_rows/<table>.jsonl failed_rows/<table>_failed.json
python bulk_ingest.py
rm checkpoints/<table>.json failed_rows/<table>.jsonl failed_rows/<table>_failed.json
python bulk_ingest.py

Or wipe all tables' state at once if you're re-loading several:

rm -rf checkpoints failed_rows
rm -rf checkpoints failed_rows

Editing the source CSV between runs is not supported. The checkpoint tracks batch indices, not row content — so if you add or modify rows in the source file and re-run, the changes can be silently skipped. For example, adding one row to a CSV that previously fit in a single batch leaves the new row in batch #0, which the checkpoint marks "skip." If you need to add or modify rows after a load has completed, post them directly to POST /api/v2/objects/{objectName} rather than re-running the script with the edited file.

Tips for multi-day runs​

  • Run from a stable host. A laptop that sleeps or moves between networks will interrupt the run; the script will resume cleanly, but you'll lose elapsed time. If you're running on a remote host over SSH, use nohup or a terminal multiplexer (tmux, screen) so the script survives the session dropping.
  • Monitor disk space. Failed-row files grow with the volume of validation failures. For a dataset with 1% failure rate across 100M rows, that's 1M JSONL entries.
  • Investigate sustained errors early. If you see persistent 5xx responses or sudden throughput drops, stop the run and submit a support ticket before exhausting retries. A dataset of this size is easier to debug at hour 2 than at hour 20.
  • Treat the failed-row files as a second-pass workload. After the bulk load completes, fix the data issues surfaced in the failed_rows/ directory and re-submit those rows in a smaller load. Read <table>_failed.json to triage what went wrong; each entry captures the row as the API saw it — extract the row payload, fix the offending field(s), and submit a small batch via POST /api/v2/objects/{objectName} directly (see the API Guide for the request shape).
  • Watch for encoding fallbacks. If the logs show a file being decoded as cp1252 or latin-1 rather than UTF-8, spot-check a few rows of that table in the warehouse for garbled characters — a mis-detected legacy encoding is the most likely cause of mojibake.

Was this page helpful?

Connect

Need help? Support

Want to learn? UiPath Academy

Have questions? UiPath Forum

Stay updated