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

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

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

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

• bq_and_s3_multi_account_dag.py — multiple accounts in parallel

License

MIT