# Bulk dataset upload

Mirror a regular flat table from your own systems into Rippit, such as an accounts table, an orders table, subscriptions, product usage or a table of custom attributes, so you can join it to your conversations, filter on it and report on it. One row in your table is one row in Rippit. It can come from anywhere you can query or export rows: a database, a data warehouse, an internal API, a spreadsheet.

You don't need to be an engineer. Copy the prompt at the bottom of this page into [Claude Code](https://claude.com/claude-code), answer its questions about where your data lives, and it writes and runs the export script for you.

To send conversations instead, such as chats, emails or calls, use the [bulk conversation upload](bulk_conversations.md). Join datasets to conversations on `conversation_id`.

---

## How it works

1. **Create the dataset once.** Give it a name, its columns and the column(s) that identify a row. Rippit returns a `dataset_id`.
2. **Ask for an upload slot** for that dataset. Rippit returns a one-time upload link.
3. **Upload one file** to that link: Parquet (recommended) or JSON Lines.
4. **Tell Rippit the upload is finished**, then **check the job** until it says `done`.

Every source works the same way: get your rows into a table, write it as Parquet and upload it. Only the first step changes, so this page has one upload script plus a few example snippets to start from.

---

## Before you start

**Get an API token.** In Rippit, click your name in the top right → **Other Settings** → **API Keys**. Tokens expire every 90 days by default; you can turn expiration off or deactivate a token manually.

**Authentication.** Send your token in the `apitoken` header on every request:

```python
headers = {'apitoken': 'YOUR_API_TOKEN'}
requests.post(url, headers=headers)
```

**Base URL:** `https://integrations.maestroqa.com/api/v2/bulk_ingest`

Bulk upload has to be switched on for your account. If it isn't, requests return `403` with "Your account does not have API ingest enabled." Ask your Rippit contact to turn it on.

---

## How datasets behave

**Primary key (`pkey_cols`).** The column or columns that identify a row, such as `account_id`. Uploading a row with a key that's already there replaces that row, so it's safe to re-run the same export every hour or every day. For an append-only dataset, like an event log, leave `pkey_cols` empty and set `"append_only_dataset": true`. Every uploaded row is kept, so re-uploading the same file creates duplicates. One of the two is required.

**Updated-at column (`updated_at_col`).** Optional. When the same key arrives more than once, the row with the later value in this column wins. Without it, the most recent upload wins.

**Columns.** You declare them when you create the dataset. If a later file has a column the dataset doesn't, it's added automatically. You can't change the primary key after the dataset is created.

**Column types:**

| Type | Use it for |
|---|---|
| `int` | Whole numbers. |
| `float` | Decimals. |
| `string` | Text and IDs. Store IDs as text even when they look like numbers. |
| `bool` | `true` / `false`. |
| `datetime` | Timestamps. Values with a timezone are converted to UTC. |
| `list<element: string>` | A list of values. The element can be any type above. |

**Column names:** letters, numbers and underscores only, starting with a letter or underscore, and at most 128 characters. Names starting with `__` are reserved. At most 500 columns.

### Parquet or JSON Lines?

Use **Parquet** when you can. Columns keep their types, files are much smaller and processing is much faster. Every column has to fit its declared type: if a `float` column holding `1.5` is declared `int`, the job fails with a message naming that column. Files that finished before the failure stay written, so fix the type and upload again.

Use **JSON Lines** (`.jsonl`, one JSON object per line) when you build rows by hand. Each line is checked on its own, and lines that don't fit come back in `rejected` with their line number while the rest are written. Text that fits the column type is accepted, for example `"12"` for an `int` or `"2026-09-27T14:02:11Z"` for a `datetime`.

```json
{"account_id": "acct_771", "plan": "enterprise", "seats": 120, "mrr": 5400.0, "is_active": true, "renewal_at": "2027-01-31T00:00:00Z"}
```

---

## The API

### 1. Create the dataset

```
POST /datasets
```

```json
{
  "name": "accounts",
  "description": "One row per customer account, synced nightly from the billing database.",
  "columns": [
    {"column": "account_id", "type": "string"},
    {"column": "plan", "type": "string"},
    {"column": "seats", "type": "int"},
    {"column": "mrr", "type": "float"},
    {"column": "is_active", "type": "bool"},
    {"column": "updated_at", "type": "datetime"}
  ],
  "pkey_cols": ["account_id"],
  "updated_at_col": "updated_at"
}
```

The response:

```json
{"dataset_id": 3394592304, "status": "created"}
```

`name` is any label up to 128 characters, such as `Accounts (billing)`. Calling it again with the same `name` returns the same `dataset_id` with `status: "exists"`, so your script can always call it first.

To see your datasets, with their `dataset_id`s:

```
GET /datasets
```

```json
[
  {
    "dataset_id": 3394592304,
    "name": "accounts",
    "description": "One row per customer account, synced nightly from the billing database.",
    "pkey_cols": ["account_id"],
    "append_only_dataset": false,
    "updated_at_col": "updated_at",
    "created_at": "2026-10-05T16:26:49Z",
    "table_url": "https://app.rippit.com/workspace/table/Y8ABK70VALW6DWGD"
  },
  {
    "dataset_id": 2718829016,
    "name": "product_events",
    "description": "Raw in-app events, appended hourly.",
    "pkey_cols": [],
    "append_only_dataset": true,
    "updated_at_col": null,
    "created_at": "2026-10-06T09:12:03Z",
    "table_url": "https://app.rippit.com/workspace/table/9CWNC8263QIIB3P8"
  }
]
```

### 2. Ask for an upload slot

```
POST /datasets/{dataset_id}/bulk?format=parquet
```

`format` is `parquet` or `jsonl`. The response is the same as for conversations:

```json
{
  "job_id": "6abc...",
  "upload_url": "https://rippit-bulk-ingest.s3.amazonaws.com/",
  "upload_fields": {"key": "...", "policy": "...", "...": "..."},
  "expires_at": "2026-09-30T15:00:00Z",
  "max_bytes": 134217728
}
```

The upload link expires after one hour.

### 3. Upload the file

Send a multipart form POST to `upload_url`. Include every entry in `upload_fields` exactly as you received it, and attach the file as `file`. Don't send your Rippit credentials on this request. A successful upload returns `204`.

```python
requests.post(up["upload_url"], data=up["upload_fields"], files={"file": open("accounts.parquet", "rb")})
```

### 4. Mark the upload finished

```
POST /jobs/{job_id}/start
```

### 5. Check the job

```
GET /jobs/{job_id}
```

Poll this every 10 to 30 seconds until `status` is `done` or `failed`.

```json
{
  "job_id": "6abc...",
  "kind": "dataset_rows",
  "status": "done",
  "result": {
    "rows_received": 50000,
    "rows_written": 49998,
    "rejected": [{"row": 17, "error": "seats: expected int64, got 'n/a'"}],
    "dataset_id": 3394592304,
    "table_url": "https://app.rippit.com/workspace/table/Y8ABK70VALW6DWGD"
  }
}
```

`row` is the 0-based line number in your file (JSON Lines only). A `failed` job has the reason in `error`.

### Edit a dataset

```
PATCH /datasets/{dataset_id}
```

```json
{
  "label": "Accounts (billing)",
  "columns": {"mrr": {"label": "MRR (USD)"}, "internal_notes": {"hidden": true}},
  "conversation_link_column": "conversation_id"
}
```

Every field is optional and fields you leave out don't change. `label` renames the table in Rippit; its `dataset_id` and `name` stay the same. `columns` relabels or hides columns. `conversation_link_column` makes that column's values clickable links to the matching conversations (see [Linking to your conversations](#linking-to-your-conversations)); setting it moves the link off any other column. The response is the dataset, as in `GET /datasets`. Unknown column names return `400`.

---

## The upload script

This works for any source. Get your rows into a [PyArrow](https://arrow.apache.org/docs/python/) table, then call `upload()`. It creates the dataset from the table's columns, splits big tables into files under 128 MB and waits for each job.

```python
# pip install requests pyarrow
import io
import os
import time

import pyarrow as pa
import pyarrow.parquet as pq
import requests

API = "https://integrations.maestroqa.com/api/v2/bulk_ingest"
HDRS = {"apitoken": os.environ["RIPPIT_API_TOKEN"]}
ROWS_PER_FILE = 1_000_000  # lower this if a file goes over 128 MB


def col_type(t: pa.DataType) -> str:
    if pa.types.is_list(t):
        return f"list<element: {col_type(t.value_type)}>"
    if pa.types.is_integer(t):
        return "int"
    if pa.types.is_decimal(t):
        return "int" if t.scale == 0 else "float"
    if pa.types.is_floating(t):
        return "float"
    if pa.types.is_boolean(t):
        return "bool"
    if pa.types.is_timestamp(t) or pa.types.is_date(t):
        return "datetime"
    return "string"


def upload(tb: pa.Table, name: str, pkey: list[str], updated_at: str | None = None, description: str = ""):
    """Empty `pkey` makes an append-only dataset."""
    tb = tb.rename_columns([c.lower() for c in tb.column_names])
    cols = [{"column": f.name, "type": col_type(f.type)} for f in tb.schema]
    body = {
        "name": name,
        "description": description,
        "columns": cols,
        "pkey_cols": pkey,
        "append_only_dataset": not pkey,
        "updated_at_col": updated_at,
    }
    r = requests.post(f"{API}/datasets", headers=HDRS, json=body)
    r.raise_for_status()
    ds = r.json()["dataset_id"]

    for i in range(0, tb.num_rows, ROWS_PER_FILE):
        buf = io.BytesIO()
        pq.write_table(tb.slice(i, ROWS_PER_FILE), buf)

        r = requests.post(f"{API}/datasets/{ds}/bulk", headers=HDRS, params={"format": "parquet"})
        r.raise_for_status()
        up = r.json()
        f = {"file": (f"{name}.parquet", buf.getvalue())}
        requests.post(up["upload_url"], data=up["upload_fields"], files=f).raise_for_status()

        r = requests.post(f"{API}/jobs/{up['job_id']}/start", headers=HDRS)
        r.raise_for_status()
        job = r.json()
        while job["status"] not in ("done", "failed"):
            time.sleep(10)
            job = requests.get(f"{API}/jobs/{up['job_id']}", headers=HDRS).json()
        print(job["status"], job.get("result") or job.get("error"))
```

---

## Example sources

These are common starting points, not a complete list. Anything that can give you rows works: if a library returns a PyArrow table, a pandas DataFrame or a list of dicts, you can turn it into a table with `pa.Table.from_pandas(df, preserve_index=False)` or `pa.Table.from_pylist(rows)` and pass it to `upload()`. Read credentials from environment variables, never from the script.

### Postgres

```python
# pip install adbc-driver-postgresql
import adbc_driver_postgresql.dbapi as pg

with pg.connect(os.environ["PG_URI"]) as conn, conn.cursor() as cur:
    cur.execute("select * from accounts")
    tb = cur.fetch_arrow_table()

upload(tb, "accounts", pkey=["account_id"], updated_at="updated_at")
```

### Snowflake

```python
# pip install "snowflake-connector-python[pandas]"
import snowflake.connector

conn = snowflake.connector.connect(
    account=os.environ["SF_ACCOUNT"],
    user=os.environ["SF_USER"],
    password=os.environ["SF_PASSWORD"],
    warehouse="ANALYTICS_WH",
    database="PROD",
    schema="CORE",
)
cur = conn.cursor()
cur.execute("select * from accounts")
tb = cur.fetch_arrow_all()

upload(tb, "accounts", pkey=["account_id"], updated_at="updated_at")
```

Snowflake returns column names in upper case; `upload()` lower-cases them, so use lower case in `pkey` and `updated_at`.

### BigQuery

```python
# pip install "google-cloud-bigquery[pyarrow]"
from google.cloud import bigquery

tb = bigquery.Client().query("select * from `proj.core.accounts`").to_arrow()

upload(tb, "accounts", pkey=["account_id"])
```

### A CSV or Excel export

```python
# pip install pandas openpyxl
import pandas as pd

df = pd.read_csv("accounts.csv")  # or pd.read_excel("accounts.xlsx")
tb = pa.Table.from_pandas(df, preserve_index=False)

upload(tb, "accounts", pkey=["account_id"])
```

To load only what changed since the last run, filter in your query, for example `where updated_at > now() - interval '1 day'`, and run the script on a schedule. Rows with a key that's already there are updated.

---

## Linking to your conversations

To join a dataset to conversations you've uploaded with the [bulk conversation upload](bulk_conversations.md), add a `conversation_id` column with the same IDs you used there, stored as `string`. You can link one row per conversation (for example ticket metadata) or many (for example the orders a customer placed during a chat). Then set it as the link column with `PATCH /datasets/{dataset_id}` and `{"conversation_link_column": "conversation_id"}`, so clicking a value opens that conversation.

---

## Limits

| Limit | Default |
|---|---|
| File size | 128 MB per upload |
| Columns | 500 per dataset |
| Uploads | 10 per minute, 100 per hour, 100 per day |

Only uploads that were actually completed count toward the limits. If you hit one you get `429` with a `Retry-After` header. Wait that many seconds and try again.

## Errors

| Code | Meaning | Fix |
|---|---|---|
| `401` | The token is missing, wrong or expired. | Check the `apitoken` header, or create a new token. |
| `403` | Bulk upload isn't enabled for your account. | Ask your Rippit contact to turn it on. |
| `404` | The dataset or job wasn't found. | Check the `dataset_id` or `job_id`. You can only see your own account's datasets and jobs. |
| `409` on `/start` | The file wasn't uploaded yet, or the job was already started. | Upload the file first, and call `/start` only once. |
| `413` | The file is over 128 MB. | Lower `ROWS_PER_FILE` and upload again. |
| `422` | The request body is invalid, such as a bad column name or type. | Read `detail` in the response for the field that failed. |
| `429` | You hit the upload limit. | Wait for the `Retry-After` time. |

---

## Paste this into Claude Code

```
I want to send a table from our systems into Rippit using its bulk dataset upload API.
Docs: https://integrations.maestroqa.com/api/v2/guides/bulk_datasets.md
API schema: https://integrations.maestroqa.com/api/v2/openapi.json

Please:
1. Ask me where the table lives (any database, warehouse, API or file), which table or query to send, and which column(s) uniquely identify a row. Ask for credentials only through environment variables.
2. Write a Python script that:
   - reads the table into a PyArrow table, starting from the closest example in the docs or any library for my source
   - uses the upload() function from the docs as written: it creates the dataset, writes Parquet files under 128 MB, uploads each one and waits for the job
   - keeps column names to letters, numbers and underscores
   - if I want it to run on a schedule, only selects rows changed since the last run
3. Run it on a small sample first (limit 1000 rows), show me the job result, then run the full export.
```