Loads

8. Loads¶

A load adds rows to one lake table. Four steps:

1. Create the load with POST /load. You get a loadId.
2. Upload one or more parquet files with POST /load/{load_id}/file, then upload each file to the link it returns.
3. Commit the load with checksums using POST /load/{load_id}/commit.
4. Poll GET /load/{load_id} until SUCCESS or FAIL.

Either all rows are added or none. Only you can see or change your loads.

8.1 Load status¶

Poll every 5 seconds while the status is PROCESSING.

8.2 POST /load: create a load¶

URL: POST https://<org>.api.comodash.io/v2/load | Scope: load:write

SDK equivalent: Load(config, load_type="APPEND_ONLY", table_name=...)

Request fields

Response 202

{"loadId": "20261013095801_355d42f6_0684_4b8f_b274_225bf7237494"}

Key errors: 400 MissingParameterError for missing fields; 400 InvalidParameterError: invalid load_type parameter (only APPEND_ONLY is accepted); 400 InvalidParameterError: table_name - Value should be in lowercase and contain only lowercase letters and underscores.

load_as_service_client_id is not verified here. If you cannot access that service client, the commit fails later with ErrorType: PermissionsError.

LOAD_ID=$(curl -s -X POST "$API/load" \
  -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" \
  -d '{"load_type":"APPEND_ONLY","table_name":"policies","load_as_service_client_id":"12"}' \
  | jq -r .loadId)

r = requests.post(f"{API}/load", headers=HEADERS, json={
    "load_type": "APPEND_ONLY", "table_name": "policies",
    "load_as_service_client_id": "12",
})
r.raise_for_status()
load_id = r.json()["loadId"]

8.3 POST /load/{load_id}/file: upload a file¶

Returns a pre-signed S3 URL and temporary credentials for uploading one parquet file. Call once per file. The credentials expire after 1 hour.

URL: POST https://<org>.api.comodash.io/v2/load/{load_id}/file | Scope: load:write

SDK equivalent: Load.upload_file(data, ...) or Load.upload_df(df, ...)

Optional request body field

Response 200

File requirements (enforced at commit time, not during upload):

• Must be a valid parquet file.
• Column names: only lowercase letters, numbers and underscores.
• If load_as_service_client_id was not given in POST /load, files must include a service_client_id column.
• If load_as_service_client_id was given, files must not include a service_client_id column.

Key errors: 400 InvalidParameterError: file_key - ... for a bad file_key; 404 NotFound: load_id does not exist for this user. for an unknown load ID.

8.4 POST /load/{load_id}/commit: commit the load¶

Starts the load pipeline. The status becomes PROCESSING. The load ID is the pipeline execution name, so calling this endpoint twice returns 400 InvalidParameter: Load Already Committed.

URL: POST https://<org>.api.comodash.io/v2/load/{load_id}/commit | Scope: load:write

SDK equivalent: Load.commit(check_sum={...})

Request body

{"check_sum": {"count(*)": 50000, "sum(face_amount)": 9876543210}}

check_sum is required and must have at least one entry. Keys are SQL aggregate expressions run over all uploaded rows. Values are what you expect. Accepts string, integer or float values.

Response 202: {"message": "Commit initiated."}

Key errors

curl -s -X POST "$API/load/$LOAD_ID/commit" \
  -H "Authorization: Bearer $TOKEN" -H "Content-Type: application/json" \
  -d '{"check_sum": {"count(*)": 50000}}'

requests.post(f"{API}/load/{load_id}/commit", headers=HEADERS,
              json={"check_sum": {"count(*)": 50000}}).raise_for_status()

8.5 GET /load/{load_id}: get load status¶

URL: GET https://<org>.api.comodash.io/v2/load/{load_id} | Scope: load:read

SDK equivalent: Load.wait_to_complete(), Load.get_load_info()

Response 200

ErrorType values

{
  "LoadStatus": "FAIL",
  "ErrorType": "CheckSumError",
  "ErrorMessages": [
    {
      "CheckResult": "FAIL",
      "ErrorType": "CheckSumError",
      "ErrorMessage": "Checksum count(*) was expecting 50001 and got 50000"
    }
  ]
}

import time

while True:
    r = requests.get(f"{API}/load/{load_id}", headers=HEADERS).json()
    print(r["LoadStatus"])
    if r["LoadStatus"] not in ("OPEN", "PROCESSING"):
        break
    time.sleep(5)
if r["LoadStatus"] != "SUCCESS":
    raise RuntimeError(r.get("ErrorMessages"))

8.6 What happens after commit (pipeline overview)¶

1. Files are collected and validated (parquet format and column names).
2. A temporary Athena table is created from all uploaded files.
3. Service client rules are enforced per row.
4. Checksums are calculated and compared with your expected values.
5. The lake table is created (if new) or columns are matched (if it exists). Column types must match.
6. Rows are inserted. Every row gets a dash$load_id column. The table is partitioned by service_client_id, dash$load_id and any extra partitions you requested.