# bulk-api-caller

Run the same Orchard ows endpoint across every row of a CSV. A JSON request
template says *what* to call; the CSV supplies the *values*; `src/bulk_runner.py`
walks the rows in a thread pool with retry-on-429.

---

## Quick map

```
configs/                        request templates (JSON) ← one per operation
data/
  prepare/                      checked-in source CSVs from the business
  input/                        runner-ready CSVs (1_prepare.py output)
  output/                       responses.json / failures.json
src/
  bulk_runner.py                Step 2 — generic runner (engine for all pipelines)
  pipelines/
    oklisten/
      1_prepare.py              Step 1 — transform AccountsDet.csv → runner CSV
      3_postprocess.py          Step 3 — responses JSON → output CSV with IDs
      schemas.py                CreateVendorSchema (int/bool coercion)
    foundation_accounts/
      1_prepare.py              Step 1 — Foundation spreadsheet (A–T) → runner CSV
      3_postprocess.py          Step 3 — responses JSON → IDs + contact block (M–S)
      schemas.py                CreateVendorSchema (int/bool coercion)
    awal_terminations/
      1_prepare.py              Step 1 — source CSV → runner CSVs (staff + delete)
      schemas.py                VendorUuidSchema (CSV shape check)
    country_attribution/
      1_prepare.py              Step 1 — source CSV → runner CSV (vendor_uuid, country_id)
      schemas.py                Marshmallow schema coercing country_id → int
    gsa_assign_cs/
      1_prepare.py              Step 1 — source CSV → runner CSV (vendor_uuid)
    cee_russia_oa_update/
      1_prepare.py              Step 1 — source CSV → runner CSV (vendor_uuid)
```

---

## OKListen pipeline

Ticket: [PLATFORM-4879](https://theorchard.atlassian.net/browse/PLATFORM-4879).

### Step 1 — Prepare input CSV

Business supplies `data/input/OKListen Phase 2 Account Integration - AccountsDet.csv`.

Transform it into a runner-ready CSV with staff IDs resolved from DB and enums mapped:

```bash
uv run python src/pipelines/oklisten/1_prepare.py \
  --input "data/input/OKListen Phase 2 Account Integration - AccountsDet.csv" \
  --output data/input/oklisten_vendors.csv
```

Requires `awsume prod` (DB lookup). Watch stderr for `TODO_*` placeholders — extend
`TIER_UUIDS`, `GENRE_IDS`, `COUNTRY_CODES`, `COMPANY_BRAND_MAP` in the script and re-run
until clean.

---

### Step 2 — Dry run (always do this first)

```bash
# Slice a small sample
head -1 data/input/oklisten_vendors.csv > data/input/oklisten_vendors-small.csv
sed -n '2,51p' data/input/oklisten_vendors.csv >> data/input/oklisten_vendors-small.csv

# Dry run against sample
uv run python src/bulk_runner.py \
  --config configs/create_vendor.json \
  --csv data/input/oklisten_vendors-small.csv \
  --dry-run
```

Validates: file paths, JSON parse, CSV load, template field coverage.
Does **not** render full payload or hit the network.

---

### Step 2 — Real run

QA, small sample first:

```bash
uv run python src/bulk_runner.py \
  --config configs/create_vendor.json \
  --csv data/input/oklisten_vendors-small.csv \
  --schema pipelines.oklisten.schemas:CreateVendorSchema \
  --base-url https://qa-ows-account.theorchard.io \
  --workers 5 --delay 0.2
```

Full run (after sample looks good):

```bash
uv run python src/bulk_runner.py \
  --config configs/create_vendor.json \
  --csv data/input/oklisten_vendors.csv \
  --schema pipelines.oklisten.schemas:CreateVendorSchema \
  --base-url https://qa-ows-account.theorchard.io \
  --workers 10 --delay 0.05
```

Outputs (written next to the CSV unless `--output-dir` given):

- `oklisten_vendors_responses.json` — successful responses by row.
- `oklisten_vendors_failures.json` — failed rows with status + body.
- `bulk_runner.log` — tee of stdout logs.

---

### Step 3 — Post-process responses → CSV

```bash
uv run python src/pipelines/oklisten/3_postprocess.py \
  data/input/oklisten_vendors_responses.json \
  --runner-csv data/input/oklisten_vendors.csv \
  --accounts-csv "data/input/OKListen Phase 2 Account Integration - AccountsDet.csv" \
  --identities-csv data/input/oklisten_identities.csv
```

Writes `oklisten_vendors_responses_ids.csv` with vendor_id, vendor_uuid, name, link,
current_subaccount_id, label contact fields merged from AccountsDet, and identity_uuid
matched on the contact email against `oklisten_identities.csv` (empty if not found or null).

---

### Step 4 — Click workstation link per vendor

Visits each `link` URL, finds the `workstation.theorchard.com` anchor, clicks it,
waits for the workstation tab to load, closes it, moves on. Drives an
already-running Chrome via CDP so your OA session is reused.

Launch a dedicated Chrome (keeps your normal profile untouched):

```bash
open -na "Google Chrome" --args \
  --remote-debugging-port=9222 \
  --user-data-dir="$HOME/chrome-oa-automation"
```

Log into oa.theorchard.com once in that window. Then:

```bash
uv run python src/pipelines/oklisten/4_click_workstation.py \
  --csv data/output/oklisten_vendors_responses_ids.csv \
  --cdp-url http://127.0.0.1:9222 \
  2>&1 | tee data/output/workstation_click_log.txt
```

Flags: `--limit N` smoke test, `--start N` resume, `--delay S` post-load sleep (default 2s).

Note: modern Chrome blocks `--remote-debugging-port` on the default profile —
you must use a separate `--user-data-dir`. Use `http://127.0.0.1:9222`, not
`localhost` (IPv6 resolution fails).

---

### Step 5 — Generate verify SQL

Emits a SQL file with vendor_ids inlined into an IN clause. LEFT JOINs
`vend_contact` against `vectorapi_access_tokens` so rows with NULL `oauth_token`
reveal vendors whose workstation click didn't mint a token.

```bash
uv run python src/pipelines/oklisten/5_gen_verify_sql.py \
  --csv data/output/oklisten_vendors_responses_ids.csv \
  --out data/output/verify_tokens.sql
```

Run the generated SQL via the MySQL client against `art_relations`.

---

## Foundation Accounts pipeline

Ticket: [PLATFORM-4980](https://theorchard.atlassian.net/browse/PLATFORM-4980).
Bulk-creates ~2.6k Foundation Media vendors via a single `POST /v2/vendors`.

Field mapping (source columns A–S → `V2CreateVendorSchema`):

| Source column | Body field | Notes |
|---|---|---|
| Account Name | `name` | |
| Company (`Foundation`) | `company_brand` (`foundation`) | |
| Owner Type | `owner` | |
| Tier (`Disregard`) | `service_tier_uuid` | mapped to `untiered` |
| Day-to-Day Contact | `assigned_to` | staff → orchadmin_users.id |
| Content Ops/Reviewer | `product_manager` | staff |
| Assigned Reviewer | `assigned_reviewer` | staff |
| D3 Y/N (`N`) | `is_distributor` | |
| Country | `country` | ISO alpha-3 |
| Genre (`Hip Hop`) | `primary_genre` | id 6 |
| Dealmaker (Closer) | `closers` | staff; sent as a one-element list |
| Label Summary | `label_summary` | |
| Main Client Contact block (M–S) | — | **no endpoint** — carried to the Step 3 output for manual follow-up |
| Currency (T) | `payment_currency` | USD fallback if blank |

`payment_currency` is read from the trailing Currency column (USD fallback if blank;
required by the API, not persisted to the vendor). Closers + product_manager ride the create payload — no
second call — which requires the matching ows-account change (`V2CreateVendorSchema.closers`;
`create_vendor` reading `product_manager`/`closers`).

### Step 1 — Prepare input CSV

```bash
uv run python src/pipelines/foundation_accounts/1_prepare.py \
  --input "data/prepare/Foundation Accounts - Filled in Orchard Spreadsheet.csv" \
  --output data/input/foundation_vendors.csv
```

Requires `awsume prod` (DB staff lookup). Watch stderr for `WARNING:` lines — extend
`TIER_UUIDS`, `GENRE_IDS`, `COMPANY_BRAND_MAP`, `COUNTRY_CODES` in the script (or
investigate unresolved staff IDs) and re-run until clean.

### Step 2 — Dry run (always do this first)

```bash
head -1 data/input/foundation_vendors.csv > data/input/foundation_vendors-small.csv
sed -n '2,51p' data/input/foundation_vendors.csv >> data/input/foundation_vendors-small.csv

uv run python src/bulk_runner.py \
  --config configs/foundation_create_vendor.json \
  --csv data/input/foundation_vendors-small.csv \
  --dry-run
```

### Step 2 — Real run

QA, small sample first, then full:

```bash
uv run python src/bulk_runner.py \
  --config configs/foundation_create_vendor.json \
  --csv data/input/foundation_vendors.csv \
  --schema pipelines.foundation_accounts.schemas:CreateVendorSchema \
  --base-url https://qa-ows-account.theorchard.io \
  --output-dir data/output/foundation_accounts \
  --workers 10 --delay 0.05
```

Outputs land in `data/output/foundation_accounts/` (per-operation folder; pass the same
`--output-dir` on every run so artifacts stay grouped).

### Reprocess failures (any run)

Every run writes `<stem>_failures.json` next to the responses. Each entry is
`{"row", "data", "status", "error"}`, where `data` is the original CSV row that failed.
**Read `status`/`error` first** — deterministic 4xx (bad data, validation, 403) will fail
again; only transient ones (429, 5xx) are worth a blind retry.

Rebuild an input CSV from the failed rows' `data`, then re-run it through the same config:

```bash
# 1. failures JSON -> retry CSV (reconstructs the original columns)
python3 -c "
import json, csv
fails = json.load(open('data/output/foundation_accounts/foundation_vendors_failures.json'))
rows = [f['data'] for f in fails]
with open('data/input/foundation_vendors-retry.csv','w',newline='') as o:
    w = csv.DictWriter(o, fieldnames=list(rows[0].keys())); w.writeheader(); w.writerows(rows)
print(len(rows), '-> data/input/foundation_vendors-retry.csv')
"

# 2. re-run just the retry CSV
uv run python src/bulk_runner.py \
  --config configs/foundation_create_vendor.json \
  --csv data/input/foundation_vendors-retry.csv \
  --schema pipelines.foundation_accounts.schemas:CreateVendorSchema \
  --base-url https://qa-ows-account.theorchard.io \
  --output-dir data/output/foundation_accounts \
  --workers 5 --delay 0.2
```

The retry has its own stem (`foundation_vendors-retry_responses.json` / `_failures.json`),
so the original run's outputs aren't overwritten. Repeat on the new `_failures.json` until
it's empty, then post-process **every** `*_responses.json` (Step 3) so the IDs CSV covers
all created accounts.

This recipe assumes failures surface as non-2xx (so they land in `_failures.json`) — true
for the Step-2 create run. The Step-4 `createIdentity` run is **GraphQL (HTTP 200 even on
errors)**, so its failures are in `*_responses.json` instead; to retry it, collect the
errored rows' `tenant_uuid` (see the Step-4 error-listing snippet) into a fresh
`tenant_uuid` CSV and re-run that through `configs/foundation_create_identity.json`.

### Step 3 — Post-process responses → CSV

```bash
uv run python src/pipelines/foundation_accounts/3_postprocess.py \
  data/output/foundation_accounts/foundation_vendors_responses.json \
  --runner-csv data/input/foundation_vendors.csv \
  --source-csv "data/prepare/Foundation Accounts - Filled in Orchard Spreadsheet.csv"
```

Writes `foundation_vendors_responses_ids.csv` with vendor_id, vendor_uuid, name, link,
and the Main Client Contact block (columns M–S) merged back by account name for manual
follow-up (those fields have no account-creation endpoint).

### Step 4 — Create the master contact per account (graphql-user `createIdentity`)

Second run. ows-permissions isn't exposed through grass in prod, so go through the GraphQL
`createIdentity` mutation (graphql-user) on the **graphql-router**, not the ows-permissions
REST endpoint. `configs/foundation_create_identity.json` creates the label's contact
identity as the **master contact** on each new account tenant.

The identity (`email`/`firstName`/`lastName`/`localization`/`masterContact`/`sendInvite`)
and `rolesToAttach` are hardcoded in the config — every Foundation row shares one contact —
so the only per-row value is `tenantUuid` (the account tenant = the Step-3 `vendor_uuid`).
`tenantType` is the GraphQL enum `ACCOUNT`. createIdentity is idempotent on email: the first
call creates the identity, later calls just add the tenant.

The run CSV needs one column — `tenant_uuid` (= the Step-3 `vendor_uuid`):

```bash
python3 -c "
import csv
with open('data/output/foundation_accounts/foundation_vendors_responses_ids.csv',newline='') as f, \
     open('data/input/foundation_tenant_access.csv','w',newline='') as o:
    r=csv.DictReader(f); w=csv.writer(o); w.writerow(['tenant_uuid'])
    for row in r:
        vu=(row.get('vendor_uuid') or '').strip()
        if vu: w.writerow([vu])
"

uv run python src/bulk_runner.py \
  --config configs/foundation_create_identity.json \
  --csv data/input/foundation_tenant_access.csv \
  --base-url https://qa-ows-grass.theorchard.io/graphql-router \
  --output-dir data/output/foundation_accounts \
  --workers 5 --delay 0.2
```

Prod base URL: `https://ows-grass.theorchard.io/graphql-router`.

Notes:
- **GraphQL returns HTTP 200 even on errors**, so a failed mutation lands in
  `<stem>_responses.json` (not `_failures.json`) with an `errors` array — the "Reprocess
  failures" recipe above won't catch these. After the run, list the rows that errored:
  ```bash
  python3 -c "
  import json
  d=json.load(open('data/output/foundation_accounts/foundation_tenant_access_responses.json'))
  bad=[e for e in d if (e.get('response') or {}).get('errors')]
  print(len(bad),'errored'); [print(e['row'], e['response']['errors'][0].get('message')) for e in bad[:20]]
  "
  ```
- The admin in `.env` must be a Settings admin with access to these account tenants. The
  router also needs `apollographql-client-name` — the runner sends it automatically.
- `sendInvite: false` → identity created in Neo4j only (no auth0 user, no invite email).
  Set `true` for a real login + invite.
- `rolesToAttach` must be roles supported by an `ACCOUNT` tenant; there is no detach on create.

---

## AWAL terminations pipeline

Ticket: [PLATFORM-4844](https://theorchard.atlassian.net/browse/PLATFORM-4844) —
set OA status to `deletion` for ~19,810 AWAL Core labels and reassign their
Product Manager + Secondary Relationship Manager to "Trust & Safety Termination"
(orchadmin_users.id `3783`).

Two endpoints, run sequentially against the same vendor list:

| Step | Endpoint | Config |
|---|---|---|
| 2a | `PATCH /v2/vendor/<vendor_uuid>/internal-staff` | `configs/awal_terminate_staff.json` |
| 2b | `DELETE /v2/vendor/<vendor_uuid>`                | `configs/awal_terminate_delete.json` |

Internal-staff payload sets `product_manager` and `quarterback_label_manager`
(= "Secondary Relationship Manager" per ows-account) both to `3783`. DELETE is
idempotent and soft-deletes by setting status to `deletion`.

### Step 1 — Prepare input CSVs

Source CSV lives at `data/prepare/AWAL Core - Terminations (revised).csv`.
Column `ABACUS Account Account ID` (= vendor_id) is the only field consumed.
We resolve each to `vendor_uuid` via `art_relations.vendor`, then write two
CSVs with identical content but different filenames (so the two bulk runs'
`_responses.json` files don't collide):

```bash
uv run python src/pipelines/awal_terminations/1_prepare.py \
  --input "data/prepare/AWAL Core - Terminations (revised).csv"
```

Outputs:
- `data/input/awal_terminate_staff.csv`
- `data/input/awal_terminate_delete.csv`

Requires `awsume prod` (DB lookup). Watch stderr for unresolved `Account ID`s —
any label without a `vendor_uuid` is dropped from both CSVs.

### Step 2a — Reassign internal staff

Dry run + small sample first (see oklisten section for the slicing idiom), then:

```bash
uv run python src/bulk_runner.py \
  --config configs/awal_terminate_staff.json \
  --csv data/input/awal_terminate_staff.csv \
  --schema pipelines.awal_terminations.schemas:VendorUuidSchema \
  --base-url https://qa-ows-account.theorchard.io \
  --workers 10 --delay 0.05
```

Only promote to prod (`https://ows-grass.theorchard.io/account`, see "Running on
prod (via grass)" below) once QA looks clean.

### Step 2b — Delete vendors

```bash
uv run python src/bulk_runner.py \
  --config configs/awal_terminate_delete.json \
  --csv data/input/awal_terminate_delete.csv \
  --schema pipelines.awal_terminations.schemas:VendorUuidSchema \
  --base-url https://qa-ows-account.theorchard.io \
  --workers 10 --delay 0.05
```

DELETE is idempotent — safe to re-run for failed rows via `--skip`.

No Step 3 post-process: raw `_responses.json` / `_failures.json` is sufficient.

---

## Country attribution pipeline

Ticket: [PLATFORM-4892](https://theorchard.atlassian.net/browse/PLATFORM-4892).

Endpoint: `PATCH /v2/vendor/<vendor_uuid>/country_id`. Body: `{"country_id": <int>}`.

Source: `data/prepare/country_attribution_040926.csv` with columns
`ACCOUNT_ID, COUNTRY_ID, ATTRIBUTED_COUNTRY`. ACCOUNT_ID is a vendor_id and must
be resolved to vendor_uuid via `art_relations.vendor`.

### Step 1 — Prepare input CSV

```bash
uv run python src/pipelines/country_attribution/1_prepare.py \
  --input data/prepare/country_attribution_040926.csv
```

Output: `data/input/country_attribution.csv` (cols: `vendor_uuid, country_id`).
Requires `awsume prod` (DB lookup). Rows with no resolvable vendor_uuid are
dropped and listed on stderr.

### Step 2 — Run

The `country_id` payload must be an **integer** (not a string), so pass the
schema to coerce it:

```bash
uv run python src/bulk_runner.py \
  --config configs/country_attribution.json \
  --csv data/input/country_attribution.csv \
  --schema pipelines.country_attribution.schemas:CountryAttributionSchema \
  --base-url https://qa-ows-account.theorchard.io \
  --workers 10 --delay 0.05
```

Dry-run + small sample first, as usual. No Step 3 — raw `_responses.json` is
sufficient.

---

## GSA Assign CS pipeline

Bulk-assign vendors so their **Relationship Manager** (`assigned_to`) is the
"Client Services" orchadmin user. Source: `data/prepare/GSA-Bulk-AssignCS.csv`
with `Account ID, Name, Relationship Manager, ...` columns. Only `Account ID`
is consumed — every row gets the same `assigned_to` value from the config.

Endpoint: `PATCH /v2/vendor/<vendor_uuid>/internal-staff`. Body: `{"assigned_to": 187}`
(orchadmin_users.id `187` = "Client Services").

### Step 1 — Prepare input CSV

```bash
uv run python src/pipelines/gsa_assign_cs/1_prepare.py \
  --input data/prepare/GSA-Bulk-AssignCS.csv
```

Output: `data/input/gsa_assign_cs.csv` (col: `vendor_uuid`).
Requires `awsume prod` (DB lookup). Rows with no resolvable vendor_uuid are
dropped and listed on stderr.

### Step 2 — Run

```bash
uv run python src/bulk_runner.py \
  --config configs/gsa_assign_cs.json \
  --csv data/input/gsa_assign_cs.csv \
  --base-url https://qa-ows-account.theorchard.io \
  --workers 10 --delay 0.05
```

Dry-run + small sample first. No Step 3 — raw `_responses.json` is sufficient.

---

## CEE & Russia OA bulk update pipeline

Ticket: [PLATFORM-4913](https://theorchard.atlassian.net/browse/PLATFORM-4913) —
align ~177 CEE & Russia accounts on the same OA configuration: Client Services
as Relationship Manager and Product Manager (orchadmin_users.id `187`),
Content Review as Assigned Reviewer (`2903`), Secondary Relationship Manager
cleared, service tier set to `diy-tier-1`, and support contact email set to
`orchardsupport@theorchard.com`.

Three endpoints, each driven from the same prepared `vendor_uuid` CSV:

| Step | Endpoint | Config |
|---|---|---|
| 2a | `PATCH /v2/vendor/<vendor_uuid>/service_tier`   | `configs/cee_russia_service_tier.json` |
| 2b | `PATCH /v2/vendor/<vendor_uuid>/internal-staff` | `configs/cee_russia_internal_staff.json` |
| 2c | `PATCH /v2/vendor/<vendor_uuid>`                | `configs/cee_russia_metadata.json` |

The internal-staff payload sends `quarterback_label_manager: null` to clear the
Secondary Relationship Manager — relies on the v2 endpoint's null-write
behavior added in PLATFORM-4913.

### Step 1 — Prepare input CSV

Source: `data/prepare/CEE & Russia Bulk.csv` (8 columns; we consume `Account ID`
and use cols C–H purely as a sanity check). The prepare script aborts if any
non-empty cell in cols C–H diverges from the constants baked into the configs —
that protects against silently bulk-setting the wrong fields if the sheet is
edited later.

```bash
uv run python src/pipelines/cee_russia_oa_update/1_prepare.py \
  --input "data/prepare/CEE & Russia Bulk.csv"
```

Output: `data/input/cee_russia_vendors.csv` (col: `vendor_uuid`).
Requires `awsume prod` (DB lookup).

### Step 2 — Run (three passes, same CSV)

Dry-run + small sample first, then in order. The three passes share one
input CSV but each writes its own `_responses.json` / `_failures.json`, so
route each pass to a distinct `--output-dir` to avoid overwrites:

```bash
for cfg in cee_russia_service_tier cee_russia_internal_staff cee_russia_metadata; do
  uv run python src/bulk_runner.py \
    --config configs/$cfg.json \
    --csv data/input/cee_russia_vendors.csv \
    --output-dir data/output/$cfg \
    --base-url https://qa-ows-account.theorchard.io \
    --workers 10 --delay 0.05
done
```

Resume an interrupted pass with `--skip N` against the same config + csv.
No Step 3 — raw `_responses.json` is sufficient.

---

## Cleanup inactive but unrevoked users pipeline

Ticket: [PLATFORM-4946](https://theorchard.atlassian.net/browse/PLATFORM-4946) —
Revokes all access on users that have been disabled by being deleted from auth0.


### Step 1 — Prepare input CSVs

Source CSV lives at `data/prepare/PLATFORM-4946_cleaned_up_users.csv`.
Some of these lines were not deactivated, so only lines where the `status` column is
`success` will be kept. It then checks the `identity_id` against neo4j to check that:

1. The `active` field is `N`
2. The `auth0userId` field is the same as `id`

This should remove all users that have been reactivated since the CSV was generated.

```bash
uv run python src/pipelines/platform-4946/1_prepare.py \
  --input "data/prepare/PLATFORM-4946_cleaned_up_users.csv" \
  --output data/output/platform-4946_identities.csv
```

Outputs:
- `data/output/platform-4946_identities.csv`

You'll need to set the Neo4j credentials in .env.

### Step 2 Revoke users

Dry run + small sample first (see oklisten section for the slicing idiom), then:

```bash
uv run python src/bulk_runner.py \
  --config configs/platform_4946_revoke_users.json \
  --csv data/output/platform-4946_identities.csv \
  --base-url https://qa-ows-grass.theorchard.io/graphql-router \
  --workers 10 --delay 0.05
```

Only promote to prod (`https://ows-grass.theorchard.io/graphql-router`, see "Running on
prod (via grass)" below) once QA looks clean.


---

## Running on prod (via grass)

The pipeline examples above target **QA** (`qa-ows-account.theorchard.io`). For the
real prod run, point `--base-url` at **grass** — Orchard's prod edge router. Grass
fronts every backend service on one host (`ows-grass.theorchard.io`) and
path-routes by prefix:

| Target | Prod base URL |
|---|---|
| ows-account | `https://ows-grass.theorchard.io/account` |
| graphql-router | `https://ows-grass.theorchard.io/graphql-router` |

`https://ows-grass.theorchard.io/account` is the `bulk_runner.py` `--base-url`
default, so omitting `--base-url` hits ows-account in prod. Same `.env` auth
headers as QA — only the JWT differs (prod `iss` = `login.theorchard.com`; see
Auth below).

---

## Calling GraphQL (via grass)

The runner is HTTP-generic, so it can drive the prod **graphql-router** too — the
Apollo Federation supergraph (graphql-user, graphql-account, …) behind grass. No
new code: a GraphQL call is just a config whose `payload` is the GraphQL envelope.

- Base URL: `--base-url https://ows-grass.theorchard.io/graphql-router`
  (QA: `https://qa-ows-grass.theorchard.io/graphql-router`).
- Config `method`: `POST`, `path`: `/graphql`.
- `payload` is `{operationName, query, variables}`. `query` is a static operation
  string; `{field}` placeholders inside `variables` fill from each CSV row exactly
  like any other config.

Auth uses the same `.env` headers as the OWS calls (`Authorization: Bearer`,
`Orchard-*`). The router additionally requires an `apollographql-client-name`
header — the runner sends `bulk-api-caller` automatically (see `build_client` in
`bulk_runner.py`), so no extra config is needed.

### Config example — `createIdentity` (graphql-user)

`configs/create_identity.json`:

```json
{
  "method": "POST",
  "path": "/graphql",
  "payload": {
    "operationName": "CreateIdentity",
    "query": "mutation CreateIdentity($identity: IdentityInput!, $tenantProfileRoles: TenantProfileRolesCreateInput!, $sendInvite: Boolean, $masterContact: Boolean) { createIdentity(identity: $identity, tenantProfileRoles: $tenantProfileRoles, sendInvite: $sendInvite, masterContact: $masterContact) { id } }",
    "variables": {
      "identity": { "email": "{email}", "firstName": "{first_name}", "lastName": "{last_name}" },
      "tenantProfileRoles": { "rolesToAttach": ["{role_id}"], "tenantType": "VENDOR", "tenantUuid": "{vendor_uuid}" },
      "sendInvite": true,
      "masterContact": false
    }
  }
}
```

`operationName` is just the client-side label for the operation; the field being
called is `createIdentity` (the SettingsV2 create mutation, owned by graphql-user).

Run it like any other config:

```bash
uv run python src/bulk_runner.py \
  --config configs/create_identity.json \
  --csv data/input/identities.csv \
  --base-url https://qa-ows-grass.theorchard.io/graphql-router \
  --workers 5 --delay 0.2
```

**GraphQL returns HTTP 200 even on errors.** A failed mutation lands in
`_responses.json` (not `_failures.json`) with an `errors` array in the body —
`bulk_runner.py` only treats transport failures (non-2xx, network) as failures.
After a GraphQL run, grep the responses JSON for `"errors"` to catch per-row
GraphQL-level failures.

---

## Auth / `.env`

Auth headers are read from `.env` at startup. Required:

```
TOKEN=<raw JWT, no "Bearer " prefix>
ORCHARD_IDENTITY_ID=<uuid>
ORCHARD_PROFILE_ID=<int>
ORCHARD_PROFILE_TYPE=<e.g. Account360Profile>
```

Optional:

```
ORCHARD_IDENTITY_UUID=<uuid>
ORCHARD_PROFILE_UUID=<uuid>
```

### How I get the JWT

I grab the token straight out of the browser session for the Account360
app — that profile holds the `account_admin` PP role I have in both QA
and prod, which is what every endpoint in this repo authorizes against.

- QA:   <https://account360.qaorch.com>
- Prod: <https://account360.theorchard.com>

Sign in, open DevTools → Network, trigger any authenticated request, and
copy the `Authorization: Bearer <jwt>` header value (without the `Bearer `
prefix) into `TOKEN`. The matching `Orchard-Identity-Id`,
`Orchard-Profile-Id`, and `Orchard-Profile-Type` headers from that same
request go into the corresponding `.env` keys.

To skip the DevTools dance, the
[sony-pde-token Chrome extension](https://github.com/theorchard/collab/tree/master/ozhovnuvatyi/sony-pde-token-chrome-ext)
exposes the current session's JWT in one click.

Token tips:

- JWT `iss` `qalogin.theorchard.com` + `aud` `workstation.qaorch.com/api` → QA.
- JWT `iss` `login.theorchard.com` → prod.
- Tokens expire (~20 min) — refresh before a long run.

---

## Request config JSON (`configs/`)

One file per operation. Placeholders `{field}` are filled from each CSV row.

```json
{
  "method": "POST",
  "path": "/v2/vendors",
  "payload": {
    "name": "{name}",
    "owner": "{owner}",
    "company_brand": "{company_brand}",
    "service_tier_uuid": "{service_tier_uuid}",
    "payment_currency": "{payment_currency}",
    "is_distributor": "{is_distributor}",
    "country": "{country}",
    "primary_genre": "{genre}",
    "label_summary": "{label_summary}",
    "assigned_to": "{assigned_to}",
    "quarterback_label_manager": "{quarterback_label_manager}",
    "assigned_reviewer": "{assigned_reviewer}"
  }
}
```

Rules:

- `method`: `GET | POST | PUT | PATCH | DELETE`.
- `path`: must start with `/`; may include `{field}` (e.g. `/v2/vendors/{vendor_uuid}`).
- Rendering: a string *exactly* equal to `"{field}"` keeps native type (int stays int,
  `"true"/"false"` → bool). A partial like `"prefix-{id}"` renders as string.
- No `headers` in the config — auth comes from `.env`.

---

## `src/bulk_runner.py` flags

| Flag | Required | Default | Meaning |
|---|---|---|---|
| `--config` | yes | — | Path to request template JSON |
| `--csv` | yes | — | Path to input CSV |
| `--base-url` | no | `https://ows-grass.theorchard.io/account` (**prod!**) | API host + service prefix (see "Running on prod (via grass)") |
| `--workers` | no | `10` | Concurrent threads |
| `--delay` | no | `0.05` | Per-worker sleep after each request (sec) |
| `--dry-run` | no | `false` | Log intent, no requests |
| `--skip` | no | `0` | Skip first N rows (resume partial run) |
| `--schema` | no | — | `module.path:Class` Marshmallow schema for row validation/coercion |
| `--consumer` | no | — | `module.path:fn` called `(row, response)` on success |
| `--output-dir` | no | CSV's dir | Where `_responses.json` / `_failures.json` land |

Rate-limiting: 429 triggers exponential backoff (`BASE_BACKOFF * 2^(attempt-1)` or
`Retry-After` header), up to `MAX_RETRIES=5` per row.

Resuming: count successful rows from the log, re-run with `--skip N` on same CSV.

---

## Adding a new bulk operation

1. Create `pipelines/<operation>/1_prepare.py` — transform source data → runner CSV.
2. Create `configs/<operation>.json` — method, path, payload template.
3. Run Step 2 (`src/bulk_runner.py`) with the new config + CSV.
4. Create `pipelines/<operation>/3_postprocess.py` if you need more than the raw responses JSON.

### Bootstrapping a pipeline with Claude Code

The repetitive parts (reading a ticket, scaffolding the prepare script,
authoring the configs, updating this README) are well-suited to a single
Claude Code prompt. The Atlassian MCP server can read the ticket but it
**cannot download xlsx attachments**, so the engineer still does the
manual step of converting the spreadsheet to CSV and dropping it in
`data/prepare/`.

Workflow:

1. Open the Jira ticket in your browser, download the xlsx attachment,
   convert it to CSV (Excel → File → Save As → CSV UTF-8), and save the
   file as `data/prepare/<source>.csv`.
2. Drop a prompt like the one below into Claude Code in this repo. The
   Atlassian MCP plugin must be authenticated (`/mcp` to verify).

````
Read PLATFORM-XXXX via the Atlassian MCP and review what fields the
business wants updated and across which accounts.

I've already downloaded the spreadsheet from the ticket, converted it to
CSV, and staged it at:
  data/prepare/<source>.csv

Build the pipeline end-to-end:

- src/pipelines/<operation>/1_prepare.py — read the CSV and resolve
  Account ID -> vendor_uuid via art_relations.vendor (DB_URL from .env,
  needs `awsume prod`). Decide column-by-column which values are
  constant for every row vs which vary per account. Write
  data/input/<operation>_vendors.csv with vendor_uuid plus one column
  per per-row value (matching the names you'll template in the
  configs). For columns whose value is the same on every row,
  sanity-check the CSV against the expected constant and abort on
  divergence — that's our guard against silently bulk-setting the
  wrong field if the sheet is edited later.
- configs/<operation>_*.json — one request template per endpoint the
  ticket asks for. Hardcode the constants directly in the payload (oa
  user IDs, service tier UUIDs, fixed emails, etc). Anything that
  varies per account stays as `{column_name}` and gets supplied by the
  prepared CSV. JSON `null` literals pass through bulk_runner
  unchanged, so use null in the payload when you need to clear a
  column. See `configs/oklisten_*.json` /
  `src/pipelines/oklisten/1_prepare.py` for a per-row-templated example
  and `configs/cee_russia_*.json` /
  `src/pipelines/cee_russia_oa_update/1_prepare.py` for an
  all-constants-plus-vendor_uuid example.
- README.md — add a new section under the existing pipeline sections,
  linking the ticket and showing the run order (one bulk_runner pass
  per config, with `--output-dir` per pass when you have more than one
  config so the response files don't collide).

Verify by running `make check` and dry-running each config against a
3-row slice of the prepared CSV. Don't commit until I review.
````

Adapt the prompt: list the exact endpoints the ticket needs, name the
operation folder, and call out which payload fields are per-row vs
constant. The two reference pipelines linked above bracket the common
cases — pure constants vs per-row variation — so picking the closer
shape and copying its structure usually saves a round of revisions.