Metadata-Version: 2.4
Name: airflow-provider-avito
Version: 0.6.0
Summary: Apache Airflow provider for Avito CPA — collect call statistics
Author-email: Michael Kozhin <michael@kozhin.cc>
License: MIT
Project-URL: Homepage, https://github.com/mkozhin/airflow-provider-avito
Project-URL: Documentation, https://github.com/mkozhin/airflow-provider-avito
Project-URL: Repository, https://github.com/mkozhin/airflow-provider-avito
Project-URL: Changelog, https://github.com/mkozhin/airflow-provider-avito/blob/main/CHANGELOG.md
Keywords: airflow,avito,provider,cpa,calls
Classifier: Framework :: Apache Airflow
Classifier: Framework :: Apache Airflow :: Provider
Classifier: Development Status :: 4 - Beta
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: Internet :: WWW/HTTP
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: apache-airflow<3.0,>=2.9.1
Requires-Dist: requests>=2.28
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Dynamic: license-file

# airflow-provider-avito

Apache Airflow provider for [Avito CPA](https://developers.avito.ru/api-catalog/cpa/documentation) — collect call statistics from the Avito advertising platform.

---

*Powered by [Claude Code](https://claude.ai/code)*

---

## Installation

```bash
pip install airflow-provider-avito
```

Requires Python 3.10+ and Airflow 2: `apache-airflow>=2.9.1,<3.0`.

## Connection

Create an Airflow connection of type **HTTP** with `conn_id = avito_default` (or any name you pass to the operator).

Authentication uses only the **Extra** field; `password` is ignored. In the single-account form `login` is read too — but only to populate the `account_id` column of output records, never for authentication.

### Single account

```json
{
  "client_id": "your_client_id",
  "client_secret": "your_client_secret"
}
```

### Multiple accounts

```json
{
  "accounts": [
    {"id": "main",    "client_id": "id1", "client_secret": "secret1"},
    {"id": "agency",  "client_id": "id2", "client_secret": "secret2"}
  ]
}
```

Use `account_id` parameter on the operator to select which account to use.

> **Note:** the `account_id` column stamped onto every output record is the record's provenance. In the multi-account form it **matches** the `account_id` selector you pass to the operator — the same value used to name file/GCS/S3 paths and BigQuery table suffixes (the example DAG uses `{BASE_DIR}/{account_id}/...` and `{BQ_TABLE}_{account_id}`). In the single-account form it is the connection's `login` (or `null` when `login` is empty).

## Quick start

```python
from airflow.decorators import dag
from airflow.models.param import Param
from airflow_provider_avito.operators.calls import AvitoCallsOperator

@dag(schedule=None, params={"date_from": Param("2026-06-01"), "date_to": Param("2026-06-07")})
def avito_calls_example():
    AvitoCallsOperator(
        task_id="collect_calls",
        avito_conn_id="avito_default",
        date_from="{{ params.date_from }}",
        date_to="{{ params.date_to }}",
        base_dir="/tmp/avito",
        output_format="json",   # or "csv"
        add_snapshot_ts=True,   # optional, see "Snapshot versioning" below
    )

avito_calls_example()
```

The operator writes one JSONL (or CSV) file per date to `{base_dir}/{safe_run_id}/{date}.json` and returns a `list[dict]` with `{"date": ..., "path": ..., "snapshot_ts": ...}` entries (`snapshot_ts` is `None` unless `add_snapshot_ts=True`).

**Plan for a slow task.** The `callsByTime` endpoint allows one request a minute. A full page is held for 62 s before the next one is asked for, and a 429 — or an error the API writes into an HTTP 200, see below — is retried at that same spacing, since the specification declares no `Retry-After` and the rate it declares is the only thing a pause can come from. A short page is followed by one more request, the one that confirms pagination is over, and that one goes out immediately. A page gives up after three retry pauses, 186 s, and the rare path through a token refresh doubles that to 372 s, because each of the two calls carries its own budget of attempts; retry pauses and the pause after a full page add up. On top of that sleep lie up to eight requests at a 30 s timeout, the token request itself and, with diagnostics on, the synchronous pushes to Loki, so **one page in the extreme case runs past ten minutes** — size `execution_timeout` for a run of many such pages (the example DAG allows two hours).

### Failures, retries and the task log

Some refusals arrive inside an HTTP 200: `{"result": {"error": {"code": 1003, "message": "…"}}}` is a timeout in Avito's own service, answered with a success status. The provider reads the error out of the body and lets the code decide:

- **Retried** — `1003`, `429`, `500`, a code the specification does not name, no code at all. The attempt waits out the rate-limit window and repeats, four attempts to a page; the last one raises. An unknown code counts as transient because the price of guessing wrong is lost minutes, not lost data.
- **Raised at once** — `1000`, `1001`, `1004`, `400`, `404`. The server refuses the request itself, and a repeat brings back the same answer.
- **Sent through a token refresh** — `1002` and `401`, the two spellings of a refused authorization. The token is refreshed and the request repeated once, exactly as after an HTTP 401, and without a pause.

**A zero is never green when it came from a failure.** A body the provider cannot read a `calls` list of dicts out of — an error object, a missing `calls`, `calls: null`, `{"calls": "abc"}`, `{"result": null}`, `{"calls": [null]}`, or a body that is not an object at all (`["a", "b"]`) — fails the task with an `AirflowException` naming what the body held instead: `no readable calls (payload_kind=dict, calls_count=1)`. A period that really holds no calls, `{"result": {"calls": []}}`, stays green and writes no file.

**Every unsuccessful attempt leaves one line in the task log**, the last one included, so a minute of waiting reads as a chronicle rather than as a silence:

```
Avito callsByTime offset=0: attempt 1/4 failed — HTTP 200, code 1003 (внутренняя ошибка Avito): не удалось получить данные о звонках: …. Retrying in 62 s
Avito callsByTime offset=0: attempt 2/4 failed — HTTP 429. Retrying in 62 s
Avito callsByTime offset=0: attempt 3/4 failed — HTTP 502. Retrying in 4 s
Avito callsByTime offset=0: attempt 4/4 failed — HTTP 200, code 1003 (внутренняя ошибка Avito): не удалось получить данные о звонках: …
```

`Retrying in N s` appears only where a pause really follows, which is how the final attempt reads as final. The line carries parsed fields only — the HTTP status, the error code with its reading, the server's `message` bounded to 300 characters and flattened onto one line, and the page's `offset`. One attempt stays one line whatever the server wrote into that message. The raw answer belongs to the other channel: it travels in the diagnostic event, if one is configured, never in the task log. The same holds for the token request, whose refusal is told by its status alone.

The other minute a run can spend in silence is the one between two full pages, and it is announced as well:

```
Avito callsByTime: full page collected, waiting 62 s before offset=1000
```

### Snapshot versioning (`add_snapshot_ts`)

By default, each DAG run writes to the same per-date path, so re-running the DAG overwrites previous output and any history of call-status changes is lost.

Set `add_snapshot_ts=True` to inject `snapshot_ts` — the DAG run's `start_date` (actual wall-clock UTC start time of the run), formatted as `YYYY-MM-DDTHH:MM:SS` — into every JSON record and into the operator's returned `snapshot_ts` key. This lets a downstream task build a unique, non-overwriting path per run (e.g. an S3 key suffixed with the snapshot timestamp) and lets ClickHouse/Spark queries pick the latest snapshot or trace status history over time:

```sql
-- ClickHouse: latest snapshot only
SELECT * FROM s3('s3://bucket/prefix/**/*.json', 'JSONEachRow')
WHERE toDateTime(snapshot_ts) = (
    SELECT MAX(toDateTime(snapshot_ts)) FROM s3('s3://bucket/prefix/**/*.json', 'JSONEachRow')
)
```

`add_snapshot_ts` only applies to `output_format="json"`; it is ignored when `output_format="csv"` (the CSV column schema is fixed).

## Output record schema

Each record contains 18 fields. The canonical ordered list of field names is also available as `CALL_FIELDS` (a `tuple[str, ...]` exported from `airflow_provider_avito.hooks.avito`) for use in downstream schema definitions or validation.

| Field | Type | Description |
|---|---|---|
| `account_id` | str \| null | The cabinet's business identifier — record provenance: the `account_id` selector (multi-account form) or the connection's `login` (single-account form); `null` when unavailable |
| `id` | int | Call ID |
| `buyer_phone` | str | Buyer phone |
| `seller_phone` | str | Seller phone |
| `virtual_phone` | str | Virtual (masked) phone |
| `create_time` | str | Creation time (RFC3339) |
| `start_time` | str | Call start time (RFC3339) |
| `date` | str | Date (YYYY-MM-DD) derived from `start_time` |
| `duration` | int | Call duration, seconds |
| `waiting_duration` | float | Wait time before answer, seconds |
| `price` | int | Price in kopecks |
| `price_rub` | float | Price in rubles (`price / 100`) |
| `status_id` | int | Status code |
| `status` | str | Status label (e.g. "Целевой") |
| `item_id` | int | Ad ID |
| `group_title` | str | Campaign name |
| `is_arbitrage_available` | bool | Whether arbitrage is available |
| `record_url` | str | Call recording URL |

When `add_snapshot_ts=True` and `output_format="json"`, a 19th field is added to every record:

| Field | Type | Description |
|---|---|---|
| `snapshot_ts` | str | DAG run's `start_date`, ISO 8601 (`YYYY-MM-DDTHH:MM:SS`). Only present when `add_snapshot_ts=True` and `output_format="json"`. |

### Call statuses

| `status_id` | `status` |
|---|---|
| 0 | Целевой |
| 1 | На модерации |
| 2 | Целевой после модерации |
| 3 | Нецелевой после модерации |

## Request diagnostics in Loki (`loki_conn_id`)

Optional, off by default. With `loki_conn_id` set, the operator emits one diagnostic event per HTTP attempt against the Avito `callsByTime` endpoint — retries and the request repeated after a refused authorization each count as an attempt — to a [Loki](https://grafana.com/docs/loki/latest/) instance. An event describes how the attempt went (severity, outcome, timing, HTTP status, the shape of the raw page), the request as it went out, and — for every attempt whose answer was not intelligible — the raw response body, so a past run can be explained afterwards in Grafana. **Read [Content policy](#content-policy) before turning this on: on an anomalous answer the response body travels as it came, and the body is treated as arbitrary sensitive data.**

Turning diagnostics on does not change the export: the same files, the same operator return value, the same exceptions with the same types and messages. A Loki outage cannot fail the task — the first push failure logs one WARNING and disables diagnostics for the rest of the run. The one cost is wall-clock: the push is synchronous, with a 2 s connect timeout and a 3 s read timeout, so an unresponsive Loki holds an attempt for about 5 s — once, before diagnostics switch themselves off. The read half bounds the quiet between received bytes rather than the whole exchange, so a Loki answering in a slow dribble can hold an attempt longer than that; only the response status is used, and the body is never downloaded. What diagnostics never absorb is the task being stopped: an `execution_timeout` firing or a SIGTERM arriving while a push is in flight interrupts the task there and then, exactly as it does with diagnostics off. A stop that arrives earlier, during the Avito request itself, cancels the push instead of being held for it, so the interrupted attempt goes unreported and the task ends as promptly as it would with diagnostics off. A stop during the pause before a retry arrives later than that attempt's push — the pause runs after it — so it leaves the event sent and prevents the next attempt.

Only the `callsByTime` requests are instrumented. A run that fails before the first page — a broken connection `extra`, a failing OAuth2 token request, an unreadable connection during `account_id` resolution — sends nothing, so the absence of events for a `dag_run` is not evidence about it: it reads the same as diagnostics being off or Loki being unreachable.

```python
AvitoCallsOperator(
    task_id="collect_calls",
    avito_conn_id="avito_default",
    loki_conn_id="loki_default",   # optional; without it nothing is sent
    date_from="{{ params.date_from }}",
    date_to="{{ params.date_to }}",
    base_dir="/tmp/avito",
)
```

### Loki connection

Create an Airflow connection with `conn_type = http`:

| Airflow UI field | Meaning |
|---|---|
| **Host** | Loki base URL, either with an explicit scheme (`https://loki.example.ru`, port allowed: `https://loki.example.ru:3100`) or a bare host (`loki.example.ru`) paired with **Schema**. An IPv6 address goes in brackets: `[::1]`, `http://[::1]:3100` |
| **Schema** | `https` or `http`. Required when **Host** carries no scheme |
| **Port** | Optional (e.g. `3100`), used only when **Host** carries neither a scheme nor a port of its own |
| **Login** / **Password** | Optional Basic Auth. Set both or neither |

The push path `/loki/api/v1/push` is appended automatically; a trailing slash on **Host** is fine, and a **Host** that already ends in the push path is taken as is.

Two configurations are equivalent: `Host = https://loki.example.ru` alone, or `Host = loki.example.ru` plus `Schema = https`.

Credentials belong in **Login**/**Password**, never in the URL: a **Host** carrying userinfo (`https://user:token@loki.example.ru`, the form Grafana Cloud publishes) is rejected with a WARNING, as are a query string and a fragment.

The scheme is never guessed. A bare **Host** with an empty **Schema** is a broken connection: diagnostics are disabled with a WARNING naming the fix, rather than silently defaulting to `http`. The same happens for an empty **Host** and for any scheme other than http/https.

Basic Auth requires HTTPS: with **Login** set and a non-HTTPS URL, nothing is sent. Half-filled credentials (**Login** without **Password**, or the reverse) count as a misconfiguration and disable diagnostics too.

Multi-tenant Loki is not supported — no `X-Scope-OrgID` header is sent. The target must be single-tenant or sit behind a gateway that stamps the tenant itself.

A push counts as delivered only on HTTP 204, the status Loki answers with. Anything else — a `200` from a reverse proxy, a redirect (redirects are not followed) — is a failure: one WARNING, and diagnostics are off for the rest of the run.

Each entry carries a single stream label, `service="airflow-provider-avito"`, so label cardinality stays constant. Everything else lives in the JSON log line and is queried with LogQL over the parsed body:

```logql
{service="airflow-provider-avito"} | json | outcome != "success"
```

Because that label is the same for every task, all tasks write into one stream. On a Loki that rejects out-of-order writes, concurrent tasks can therefore have a push refused with a 4xx, which disables diagnostics for that task.

Outside an operator, the same client can be handed to the hook directly: `AvitoHook(avito_conn_id=..., loki=LokiClient(conn_id="loki_default", context={...}))`.

### Event fields

| Field | Description |
|---|---|
| `schema_version` | Event format version, currently `2` |
| `dag_id`, `task_id`, `dag_run_id`, `try_number`, `map_index` | Correlation with the Airflow task instance (`map_index` is `-1` when not mapped). These five are stamped by the Loki client at push time; the other fields come from the request itself |
| `outcome` | How the attempt ended — see the table below |
| `level` | Severity of the attempt: `info`, `warn` or `error` — see [Severity](#severity-level) below |
| `account_id` | The cabinet whose calls are being collected, as stamped onto output records |
| `offset`, `date_time_from` | Request parameters of the paginated page |
| `attempt`, `max_attempts` | Retry counters for one page request: `attempt` counts from 1 up to `max_attempts` as 429/5xx responses and retryable error codes inside an HTTP-200 body are retried, both out of the same budget. The request repeated after a refused authorization starts its own count, and a refusal can arrive on any attempt of either call, so `after_token_refresh` — not the counter — is what tells the two apart |
| `after_token_refresh` | `true` on the request repeated after a refused authorization — HTTP 401, or the codes `1002` / `401` inside an HTTP-200 body — `false` on the first try of the page. It answers "which of the two calls is this", not "how did it end": both a refresh that worked and one that did not carry `true` on the second event |
| `sent_at` | UTC ISO 8601 timestamp taken just before the request is sent |
| `request_method`, `request_url` | `"POST"` and the `callsByTime` endpoint |
| `request_headers` | The headers the provider sets — `Authorization` (masked, see below), `X-Source`, `Content-Type` |
| `request_body` | Copy of the JSON body sent: `dateTimeFrom`, `limit`, `offset` |
| `duration_ms` | Wall-clock duration of the HTTP attempt |
| `http_status` | Response status, `null` when the request never got one |
| `calls_count` | Number of entries in the raw page as the API returned it, before the export narrows them to the period, `null` when no `calls` list was recognised |
| `calls_shape_ok` | Whether the page held a `calls` list of dicts |
| `payload_kind` | Which shape the body turned out to have: `dict` (`calls` is there, holding the list it promises or an empty value of another type), `calls_absent` (no `calls` key at all), `calls_non_list` (`calls` holds a non-empty value that is not a list), `result_non_dict` (`result` holds something other than a dict), `non_dict` (the body itself is not a dict) |
| `error_code`, `error_message` | `code` and `message` of the refusal the answer carried, in whichever of the three shapes it came: `{"result": {"error": {…}}}`, `{"error": {…}}` or `{"result": {"code": …, "message": …}}` — the last one has no `error` object at all. The message is bounded to 300 characters and flattened onto one line. Filled for an HTTP-200 body, for an HTTP 401 and for an `http_error` status whose body names an error; a retried 429 or 5xx is told by its status alone — its body is never parsed, so both stay `null`. `error_code` also decides what the provider does next, so it is the field to group a dashboard by |
| `exception_type`, `exception_message` | Type of the exception that ended the attempt; the message is filled only for a JSON parse error reported by the standard decoder, from a fixed vocabulary |
| `rate_limit_limit`, `rate_limit_remaining` | `X-RateLimit-*` headers, collected on HTTP 429 |
| `response_body` | Raw response text, bounded and with the live token cut out — see [Raw response body](#raw-response-body) below |

`calls_count`, `calls_shape_ok` and `payload_kind` all stay `null` for any attempt that never produced a parsed HTTP-200 body.

The request bounds only the start of the period: the body carries `dateTimeFrom`, and the end of it is applied by the export itself, which keeps a record only when its date falls in the requested range and stops as soon as a page holds nothing but records past it. So `calls_count` counts what the API answered with, not what reached the file, and a run that collected nothing while `calls_count > 0` is the ordinary shape of "the period is empty, later days are not". How many records the export kept is in the task log, not in the event: an event describes one HTTP attempt, and the export narrows across all of them.

#### The request as it went out

`request_method`, `request_url`, `request_headers` and `request_body` together are a template of the request, not a literal transcript of the wire. Two things separate them:

- **`Authorization` carries a mask**, `"Bearer eyJhbG…c4f2"` — the `Bearer ` prefix kept, the token reduced to its first six and last four characters, joined by `…`. A token shorter than twenty characters — twice what the mask shows — is replaced whole by `***`, so the mask never spells out most of the value. The mask is enough to tell one token from another; replaying the request means substituting a live one.
- **Only the headers the provider sets are listed.** The ones `requests` adds for the connection — `Accept`, `User-Agent`, `Content-Length`, `Connection` — are not in the event: they do not change what the request means.

`request_headers` and `request_body` are nested objects. In LogQL, `| json` flattens nesting with an underscore, which is how these fields are queried:

```logql
{service="airflow-provider-avito"} | json | level = "error"
{service="airflow-provider-avito"} | json | outcome = "empty_shape" and request_body_offset > 0
{service="airflow-provider-avito"} | json | line_format "{{.request_headers_Authorization}} {{.response_body}}"
```

### `outcome` values

| Value | Meaning |
|---|---|
| `success` | HTTP 200 with a well-formed page, including an empty one — "no calls in this period" is a valid answer |
| `empty_shape` | HTTP 200 in which no `calls` list of dicts was recognised — the attempt raises, whatever `payload_kind` says about it |
| `api_error` | HTTP 200 carrying an `error` object. The code decides: `1000`, `1001`, `1004`, `400` and `404` raise at once; anything else — `1003`, `429`, `500`, an unnamed code, no code at all — is retried at the rate-limit spacing of 62 s and raises on the last attempt |
| `auth_error` | HTTP 401, or the codes `1002` / `401` inside an HTTP-200 body — the token is refreshed and the request repeated once, with no pause |
| `retryable_error` | HTTP 429, 500, 502, 503 or 504 — retried, or, on the last attempt, raised. A 429 waits out the rate-limit window — 62 s, the declared minute plus margin; the other statuses walk a short backoff ladder indexed by attempt number, 1/2/4 s. Any other 5xx (`501`, `505`, …) is an `http_error` |
| `http_error` | Any other non-200 status |
| `network_error` | The request never completed (timeout, DNS, TLS, proxy) |
| `invalid_json` | HTTP 200 whose body could not be parsed |
| `unexpected_error` | A body that is valid JSON but not an object (`["a", "b"]`) — the attempt raises, naming `payload_kind=non_dict` — and, as a safety net, an attempt that ended some other way |

Every `empty_shape` ends the attempt with an `AirflowException`; `payload_kind` says what the body held instead of a page:

- `calls_absent` — no `calls` key at all (`{}`, `{"result": {}}`).
- `dict` — the key is there, holding either an empty value of another type (`calls: null`, `""`, `{}`), and then `calls_count` is `null`; or a list holding non-dicts (`{"calls": [null]}`), and then `calls_count` counts them. The exception text carries the count for the same reason: it is what separates the two.
- `calls_non_list` — a non-empty `calls` that is not a list (`{"calls": "abc"}`, `{"calls": {"1": {…}}}`).
- `result_non_dict` — `result` holds something other than a dict (`{"result": null}`, `{"result": []}`).

The one body the export goes on reading is a `calls` that is a list of dicts, the empty list of a period without calls included — and that one is `success`, not `empty_shape`.

`offset` says how much the failure cost. At `offset = 0` nothing had been collected yet; at `offset > 0` pagination broke in the middle, and the records gathered before it go down with the run instead of into a partial file. Either way the task ends red, which is the signal to alert on: every failure ends the run, so the red task is the alert and a file on disk means a complete export.

### Severity (`level`)

| `level` | When | Meaning |
|---|---|---|
| `info` | `success` with `calls_count > 0`; `success` with `calls_count = 0` at `offset > 0` | The answer is intelligible: either there are records, or this is pagination ending |
| `warn` | `success` with `calls_count = 0` at `offset = 0`; `auth_error` **before** the token refresh; `retryable_error` with an attempt still left; `api_error` with an attempt still left and a code outside the fatal set | Nothing came back for the period; or a situation that fixes itself — an expired token, a retry |
| `error` | `auth_error` **after** the token refresh; `empty_shape`; `api_error` with a fatal code or on the last attempt; `http_error`; `network_error`; `invalid_json`; `unexpected_error`; `retryable_error` on the last attempt | The answer is unintelligible, access did not come back, or the request never completed |

`level` answers "is the answer intelligible, and is there still hope", not "did the task fail". Hence `empty_shape` is an `error` at **any** `offset`: a body with no `calls` list in it is exactly "something unintelligible arrived". An `api_error` is read the same way, with the code as the second question: a retryable code with an attempt left still has a repeat ahead of it, a fatal one has nothing ahead of it at any attempt.

`offset` splits `info` from `warn` only for a **valid** empty page (`success` with `calls_count = 0`): on the first page that means the period holds no calls at all, further on it is how pagination ends. `after_token_refresh` splits the two refusals — whether they arrived as a 401 or as a code in a 200 body: the first is a token expiring, the second means access did not come back.

`warn` at `offset = 0` is therefore narrower than "the export came out empty": it fires when the API itself had nothing to give, while an answer holding only records past the end of the period is `info`. An alert on an empty export belongs on the number of records a run collected.

**This table is also the body export policy.** The same `level` decides both the severity shown in Grafana and whether the raw body leaves the process (see below). Moving a row here changes what content is shipped, not just how alerts are coloured.

### Raw response body

`response_body` holds the response text, bounded to 32768 characters (`_BODY_LIMIT`, fixed in the provider — there is no connection or operator setting for it). A longer body is cut to that budget and ends with `…[truncated]`. The text is read with the charset the server named in `Content-Type` and as UTF-8 when it named none. The live bearer token is replaced with `<token>` wherever the answer echoes it back; an answer that spells the token out in an encoding it never named — UTF-16 read as UTF-8 — is dropped whole instead, so `response_body` is `null` there.

| Situation | `response_body` |
|---|---|
| `level = "info"` | `null` — diagnostics deliberately do not read the body |
| Any other level, with a response whose bytes could be read | The response text, bounded to `_BODY_LIMIT`, token cut out |
| `network_error` — there is no response | `null` |
| A response exists but its bytes could not be read, or spell the token out in an encoding the answer never named | `null` |
| Diagnostics off (`loki_conn_id` unset) or the task being stopped | `null` — the body is not read and nothing is pushed |

The key is always present; `null` in it is not a promise that the level was `info`. Not reading a body saves the decode and the copy the event would carry, not the transfer: `requests` has the whole answer in memory by the time the attempt is classified, and the branches that tell a refusal in parsed terms — an HTTP 200, an HTTP 401 and any other non-200 outside the retry set — parse it either way. A 429 or a 5xx builds its message from the status alone.

In a healthy run every event is `info`, so no body travels at all. Volume grows on empty and failing runs — and on retries: a `retryable_error` with an attempt left is a `warn`, as is a retryable `api_error` with an attempt left, so **every** unsuccessful try ships its body: four of them when a storm exhausts a page's attempts — the three `warn` tries and the `error` that ends them — and twice that when a token refresh puts a second budget of attempts behind the same page. Retry events for a 429 stand a minute apart from one another, because the pause is derived from the request rate the API declares — the operation's description allows one request a minute — rather than from the answer, for which the specification declares neither a `Retry-After` header nor a reset time.

One event is one Loki line, and Loki refuses a line longer than `limits_config.max_line_size` — 256 KB by default. With every bounded field at its budget and a body of control characters, the widest they get once JSON has escaped them, a line measures about 200 KB, so the default leaves room. An installation that lowered that limit — Grafana Cloud, a tuned self-hosted Loki — answers such a push with a non-204, and that first refusal disables diagnostics for the rest of the run, exactly as any other push failure does. Check `limits_config.max_line_size` on your instance before turning this on.

### Content policy

The raw response body leaves the process whenever the answer was not intelligible — that is, at every `level` other than `info` (see the tables above). **Treat it as arbitrary sensitive data.**

- **A body without recognised records is not a body without PII.** A response can carry phone numbers, `recordUrl` links and secrets inside an `error.details` object while holding no `calls` list at all — and that body ships whole.
- **Known edge:** an anomalous outcome alongside recognised records — `{"calls": [ …78 records… ], "error": {…}}`, or a `calls` list with mixed elements — ships a body containing buyer phones, virtual numbers and recording URLs.
- **The token guarantee covers the two channels the provider controls.** The bearer token is masked in `request_headers` and cut out of `response_body`. Fields derived from the server's own answer — `error_message` and `rate_limit_*` — are size-limited but not redacted: if the API writes the token into `error.message`, it lands in the event. That is a deliberate boundary, not an oversight — such a message answers a request made with that same token, which by then has usually been rejected. The boundary covers the task log too: the line an unsuccessful attempt leaves and the text of the exception that ends the page carry that same `message`, word for word.
- **The structured fields are a structure, not a boundary.** `error_code`/`error_message` and `exception_message` are narrow, queryable summaries, and the rules below describe exactly what goes into them; they describe the failure, they do not bound what the event discloses, because the body travels alongside.
- **There is no setting that keeps diagnostics on and bodies out.** The level table decides what travels; the only way to stop bodies from leaving is to leave `loki_conn_id` unset, which turns the whole feature off.
- Response headers other than the two `X-RateLimit-*` are not copied, and the text of a network exception — proxy URL with credentials included — never reaches the event.
- Retention and access follow Loki: bodies live as long as the instance keeps them, and everyone holding the shared Loki credentials can read them.

How the structured fields are built:

- From `error` only `code` (an `int`) and `message` (a value whose type is exactly `str`, flattened onto one line and truncated to 300 characters; a message that flattens to nothing is reported as no message) are taken. A value of an unexpected type is described by its type — `<non-dict error: list>`, `<non-str message: dict>` — rather than serialised, so nested keys such as `details` or `trace` are not summarised into the event.
- `exception_message` is filled only for `invalid_json`, and only when the standard JSON decoder reported the failure. It is rebuilt from the exception's own attributes rather than from its rendered text, and the wording is chosen from a fixed vocabulary of the decoder's own literals — `Expecting value`, `Expecting ',' delimiter`, `Expecting ':' delimiter`, `Expecting property name enclosed in double quotes`, `Extra data`, `Unterminated string starting at`, `Invalid control character at`, `Invalid \escape`, `Invalid \uXXXX escape` — followed by the position counted in the document: `Expecting value: line 1 column 1 (char 0)`. Anything the decoder words differently is reported as `<other decoder message>` with the same position, which keeps the field a fixed vocabulary: some decoder messages are formatted around a character taken from the document (the pure-Python scanner writes `Invalid \escape: 'q'`), and the field stays a description of the failure rather than a quotation of the answer — the answer itself travels in `response_body`, where it is bounded and the token is cut out. A parse failure of any other origin — a third-party decoder, a response object of unknown provenance — records `exception_type` alone, as does every other outcome: those exception texts render whatever was in flight, and for a network failure that is the environment's proxy URL, credentials included.
- Of the response headers, only the two `X-RateLimit-*` are copied, and only when their type is exactly `str`, truncated to 32 characters. A value of any other type is described by its type (`<non-str header: int>`), so no unknown object is ever rendered into the event.
- Truncation bounds length, not content. `error_message` is free text written by the API, so it is size-limited but not redacted — that is the honest edge of the guarantee.

## Examples

Full production examples with BigQuery + S3 upload are in [`examples/`](examples/):

- [`bq_and_s3_multi_account_dag.py`](examples/bq_and_s3_multi_account_dag.py) — multiple accounts in parallel

## License

MIT
