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

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

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 is retried at that same spacing — the specification declares no Retry-After, so the pause comes from the rate it declares instead. 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. Those figures are sleep alone, with the HTTP requests on top, so treat them as the lower bound when sizing execution_timeout (the example DAG allows two hours).

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

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.

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:

{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

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.

offset says which of two failures an empty_shape is. At offset = 0 the whole period exported as zero calls. At offset > 0 pagination broke in the middle: the export reads the unrecognised body as an empty page, stops there and writes the records it had already collected, so the file exists, the task is green, and the data is incomplete. That run should be re-exported — and it is the case worth alerting on, because unlike a zero result it looks like a successful run.

Severity (level)

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, including the quiet half where the task ends green: a body with no calls list in it is exactly "something unintelligible arrived".

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 401s: 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.

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 on a non-200 status its first 200 characters go into the exception message either way.

In a healthy run every event is info, so no body travels at all. Volume grows on empty and failing runs — and on retries: retryable_error with an attempt left is a warn, so every unsuccessful 429/5xx 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 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, 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 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/:

• bq_and_s3_multi_account_dag.py — multiple accounts in parallel

License

MIT