airflow-provider-avito

Apache Airflow provider for Avito CPA — collect call statistics from the Avito advertising platform.

Powered by Claude Code

Installation

pip install airflow-provider-avito

Requires Python 3.10+ and apache-airflow>=2.9.1.

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

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

Multiple accounts

{
  "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

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

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:

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

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

Call statuses

Request diagnostics in Loki (loki_conn_id)

Optional, off by default. With loki_conn_id set, the operator emits one compact diagnostic event per HTTP attempt against the Avito callsByTime endpoint — retries and the request repeated after a 401 token refresh each count as an attempt — to a Loki instance. The events describe how each attempt went (outcome, timing, request parameters, HTTP status, the shape of the raw page), so a past run can be explained afterwards in Grafana. No call data is sent.

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 or the backoff pause — 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.

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.

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:

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:

{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

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

outcome values

empty_shape splits in two, and the other fields say which half an event belongs to:

• payload_kind = "dict" with calls_count = null — a body with no calls key, calls: null, or another empty calls value ("", {}). The export treats it as an empty page: green task, no file.
• payload_kind = "calls_non_list" or "result_non_dict", or calls_shape_ok = false with a non-null calls_count — a non-empty calls that is not a list ({"calls": "abc"}, {"calls": {"1": {…}}}), a non-dict result ({"result": null}, {"result": []}), or a calls list holding non-dicts ({"calls": [null]}). The export fails downstream with AttributeError or TypeError: the task ends red.

The quiet half — empty_shape with payload_kind = "dict" and calls_count = null — and success with calls_count = 0 are the pair worth watching: both end with a green task and no file, and only the event tells them apart.

Content policy

The event is assembled from the fixed allowlist above. The raw response body, response headers other than the two X-RateLimit-*, call records, and the error object as a whole never leave the process.

• From error only code (an int) and message (a value whose type is exactly str, truncated to 300 characters) 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 cannot ride along.
• 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, because some decoder messages are formatted around a character taken from the document (the pure-Python scanner writes Invalid \escape: 'q'), and a single character of the body is still the body. A parse failure of any other origin — a third-party decoder, a response object of unknown provenance — may quote what it was parsing and records exception_type alone, as does every other outcome: those exception texts embed the response body or, for network failures, 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/:

• bq_and_s3_multi_account_dag.py — multiple accounts in parallel

License

MIT