airflow-provider-cian 0.5.1

Apache Airflow provider for Cian.ru Builder API — collect calls and chats statistics

airflow-provider-cian

---

Powered by Claude Code

---

Airflow provider for Cian.ru Builder API — collect calls and chats statistics.

Installation

pip install airflow-provider-cian

Requirements: Python 3.10+, Apache Airflow 2.9.1–2.x.

Connection Setup

Single account

Create an HTTP connection in Airflow (Admin → Connections):

Field	Value
Connection Id	cian_default (or any name)
Connection Type	HTTP
Host	https://public-api.cian.ru
Login	Account ID — a label for this cabinet (required)
Password	Bearer token from your Cian Builder cabinet

The provider reads conn.host as base URL and conn.password as Bearer token. Login is required: it is written into every output record as account_id and into the file path as the cabinet segment. Without it the task fails with AirflowException before any API request. The value is a label you choose (Cian's API does not expose a real cabinet ID); a good choice is the legal entity's INN or name from your contract.

Multiple accounts

To collect data from several Cian cabinets using a single connection, put tokens in the Extra field as JSON:

{
  "accounts": [
    {"id": "111", "token": "Bearer <token-for-cabinet-111>"},
    {"id": "222", "token": "Bearer <token-for-cabinet-222>"}
  ]
}

The id is a label you choose that uniquely identifies the cabinet — Cian's API does not expose a real cabinet ID, so this is not a value from Cian (a good choice is the legal entity's INN or name from your contract). Non-alphanumeric characters are sanitised ([^\w-] → _) uniformly wherever the id is used: file paths, BigQuery table names, token lookup and the account_id field written into every output record.

Token resolution

The token source is controlled solely by account_id on the operator — not by whether Password or Extra is filled in:

account_id on operator	Token taken from
not set (None)	conn.password — error if empty
"111"	extra.accounts entry where id == "111" — error if not found or token empty

Password and Extra are independent fields and do not interfere with each other. If both are filled in, the operator uses one and ignores the other depending on account_id.

Operator Parameters

CianBuilderReportsOperator:

Parameter	Type	Default	Description
cian_conn_id	str	cian_default	Airflow connection ID
date	str	required	Collection date, YYYY-MM-DD. Supports {{ ds }} template. Validated at run time — a non-YYYY-MM-DD or calendar-invalid value raises AirflowException before any API call
base_dir	str	/tmp/cian	Base directory for output files
output_format	str	json	json (JSONL) or csv
account_id	str | None	None	Cabinet ID for multi-account mode (matches id in Extra JSON). Also written (sanitized) into the account_id field of every output record
add_snapshot_ts	bool	False	Add snapshot_ts field (naive-UTC run start time, ISO 8601) to each JSON record. Ignored for output_format='csv'.

Return value (execute contract)

For a day with data, the operator writes the file and returns a self-describing dict, pushed to the return_value XCom:

{"date": "2026-07-01", "path": "/tmp/cian/<account_id>/<run_id>/2026-07-01.json"}

For an empty day (Cian returned reports: []), the operator writes no file and returns None. Airflow does not push an XCom for None, so an empty day simply drops out of the collect mapped task's result list — no downstream mapped instance is created for it.

Breaking change (unreleased): execute() previously returned the path as a plain str. It now returns dict | None. DAGs (or XCom consumers) that read the collect result as a string must unwrap item["path"].

DAG author warning — items or []: any aggregator that consumes the collected list MUST start with items = list(items or []). When the entire period is empty, no mapped instance writes an XCom and Airflow hands the aggregator None, not [] — a naive len(items) would raise TypeError.

Fully empty period. In the v1 and multi-account DAGs (bq_and_s3_dag.py, bq_and_s3_multi_account_dag.py), which have separate mapped upload/load tasks, the aggregators return [], those mapped tasks expand to zero instances and are marked skipped, and the dag_run still succeeds. In the v2 DAG (bq_and_s3_dag_v2.py) the uploads happen inside a single process_date task, so an empty day is just process_date returning None in state success — there is no skipped task, even when the whole period is empty.

Broken API response. A 200 response without result.reports (or with a non-list there) now fails the task with AirflowException instead of being silently treated as an empty day.

The output file path always contains the cabinet segment (the sanitized Account ID):

account_id	conn.login	Path
not set	not set	error — AirflowException (Account ID is required)
not set	"123"	{base_dir}/123/{run_id}/{date}.{ext}
"123"	any	{base_dir}/123/{run_id}/{date}.{ext}

In single-account mode, Login on the connection is required and supplies the Account ID for both the file path and the account_id field. If neither account_id nor Login is set, the task fails with AirflowException before any API request.

Output Schema

Base schema (19 fields) — present in all records regardless of format:

id, account_id, newbuilding_id, newbuilding_name, date, datetime, action_type, searcher_phone, searcher_ct_phone, builder_user_ct_phone, builder_user_phone, builder_sip_uri, call_duration, tariff_price, auction_bet, cashback_spent, billing_price, has_claim, is_targeted

• account_id — the sanitized Account ID of the cabinet this record belongs to (second field, right after id); same value as the file-path segment. Lets you tell records from different cabinets apart in the data itself, not only from the file path
• date — collection date (YYYY-MM-DD), always equals the operator's date parameter; safe for BigQuery date partitioning
• datetime — original API datetime with explicit Moscow offset (YYYY-MM-DDTHH:MM:SS+03:00); sub-second digits returned by the API are dropped (truncated, not rounded), so the value always has second precision
• is_targeted is computed: billing_price > 0.

When add_snapshot_ts=True and output_format='json', each record also contains a 20th field:

• snapshot_ts — dag_run.start_date formatted as YYYY-MM-DDTHH:MM:SS (naive UTC, no timezone offset). All records within a single DAG run share the same value.

Snapshot Versioning

The billing_price and is_targeted fields can change retroactively after initial collection (Cian may adjust billing post-factum). To track these changes over time, enable add_snapshot_ts=True:

collect = CianBuilderReportsOperator.partial(
    task_id="collect",
    cian_conn_id="cian_default",
    output_format="json",
    add_snapshot_ts=True,
).expand(date=dates)

Each JSON record will include snapshot_ts — the real wall-clock start time of the DAG run (dag_run.start_date, naive UTC). All records within the same run share one timestamp.

To query all records from the latest snapshot in ClickHouse:

SELECT *
FROM cian_calls
WHERE snapshot_ts = (
    SELECT max(snapshot_ts) FROM cian_calls
)

Or to build a history of billing_price changes:

SELECT id, billing_price, snapshot_ts
FROM cian_calls
ORDER BY id, snapshot_ts

BigQuery note: the base BQ_SCHEMA in the v1 and v2 examples (bq_and_s3_dag.py, bq_and_s3_dag_v2.py) is fixed at 19 fields (including account_id). The multi-account example (bq_and_s3_multi_account_dag.py) uses add_snapshot_ts=True, so its schema has 20 fields (base 19 + snapshot_ts). When enabling add_snapshot_ts=True on a 19-field schema, either add a snapshot_ts STRING column or use ignore_unknown_values=True on the load job — otherwise BigQuery will reject records with the extra field.

JSON only: add_snapshot_ts=True has no effect when output_format='csv'. The CSV schema remains 19 fields.

Multi-Account Support

list_accounts() reads the connection Extra JSON and returns a list of Account objects. Use it at DAG parse time to build one TaskGroup per cabinet:

from airflow_provider_cian.accounts import Account, list_accounts

accounts = list_accounts("cian_default")  # returns [] if no accounts configured
for account in accounts:
    with TaskGroup(group_id=f"cabinet_{account.id}"):
        CianBuilderReportsOperator.partial(
            task_id="collect",
            cian_conn_id="cian_default",
            account_id=account.id,   # routes to the correct token
            ...
        ).expand(date=dates)

See examples/bq_and_s3_multi_account_dag.py for a full working example with GCS, BigQuery and S3 uploads.

Internal resolution functions

airflow_provider_cian.accounts also exposes two lower-level helpers used by the hook and operator. DAG authors typically do not need these directly:

• resolve_cabinet_id(conn_id, account_id) — returns the sanitized cabinet id for an operation. In multi-account mode (account_id is set) it returns sanitize_id(account_id) without reading the connection. In single-account mode it reads conn.login lazily and returns its sanitized form (or None if unset). Sanitization is idempotent, so a value already coming from list_accounts() is unchanged.
• resolve_token(conn, account_id) — returns the authentication token. In multi-account mode it finds the entry in conn.extra.accounts whose sanitized id matches the sanitized account_id (canonical lookup key), so a raw account_id like "a.b" matches an entry stored as "a.b" or "a_b". If two or more entries collapse to the same sanitized id (an ambiguous collision), it raises AirflowException rather than silently picking one — a deliberate asymmetry with list_accounts(), which deduplicates such collisions with a WARNING and keeps the first. In single-account mode it returns conn.password. Raises AirflowException when the token cannot be resolved.

Example DAG

from datetime import date, timedelta
from airflow.decorators import dag, task
from airflow.operators.python import PythonOperator
from airflow_provider_cian.operators.builder_reports import CianBuilderReportsOperator
import os

@dag(schedule=None, catchup=False, max_active_tasks=3)
def cian_reports():
    @task
    def get_dates():
        yesterday = date.today() - timedelta(days=1)
        return [(yesterday - timedelta(days=i)).isoformat() for i in range(7)]

    dates = get_dates()

    collect = CianBuilderReportsOperator.partial(
        task_id="collect",
        cian_conn_id="cian_default",
        base_dir="/tmp/cian",
        output_format="json",
    )
    collected = collect.expand(date=dates).output  # list of {"date", "path"} — only days with data

    # Route uploads through an aggregator that turns collected items into expand kwargs.
    # `items or []` is mandatory: for a fully empty period Airflow hands the aggregator
    # None (no XCom was written), not [].
    @task
    def to_s3_params(items):
        items = list(items or [])
        return [
            {"filename": it["path"], "dest_key": f"cian/{it['date']}.json"}
            for it in items
        ]

    # Wire the upload here (left commented — this is a template stub):
    # LocalFilesystemToS3Operator.partial(...).expand_kwargs(to_s3_params(collected))

    def cleanup(ti, **ctx):
        # xcom_pull on a mapped task returns a list-like of the per-day
        # {"date", "path"} dicts (only days with data), or None when the whole
        # period is empty — never a bare dict.
        items = ti.xcom_pull(task_ids="collect")
        for item in (items or []):
            path = item["path"]
            if path and os.path.exists(path):
                os.remove(path)

    collected >> PythonOperator(task_id="cleanup", python_callable=cleanup, trigger_rule="all_done")

cian_reports()

Rate Limiting

The API limit is ≤10 req/s per token (per Cian account). The hook adds a 100ms sleep before each request. max_active_tasks=3 on the DAG level provides additional safety margin.

If multiple clients share the same IP and you still get 429 errors, create an Airflow Pool:

airflow pools set cian_api 5 "Cian API rate limit pool"

Then pass pool="cian_api" to CianBuilderReportsOperator.partial(...).

Error Handling

CianNotFoundError (subclass of AirflowException) is raised when the API returns a "not found" response for a resource. get_newbuilding_name catches it internally and returns "Неизвестно" — DAG authors don't need to handle it for that method. For custom hook usage:

from airflow_provider_cian.hooks import CianNotFoundError

A record whose date field is missing or is not a string fails the task with AirflowException naming the record id. A date string the stdlib cannot parse fails the same way — AirflowException: Record id=<id> has an unparsable 'date': '<value>' — where <value> is quoted exactly as the API sent it, before any normalisation (Z → +00:00, sub-second digits dropped). That keeps the failing precision or format readable straight from the log, and an unreadable timestamp surfaces loudly instead of being silently "repaired" (see docs/adr/0005-datetime-drop-subsecond.md).

Retry Behaviour

On HTTP 429 or 5xx: exponential backoff — 1s, 2s, 4s (3 attempts total), then AirflowException.

License

MIT