Напиши DAG для загрузки данных через REST API

Создайте DAG с задачей, которая читает API через connection, проверяет ответ и идемпотентно записывает данные. Секреты не должны находиться в коде DAG.

Минимальный пример для Airflow 3 с HTTP- и PostgreSQL-провайдерами. Предполагаются небольшой JSON-массив объектов с id и готовая таблица api_items(id text PRIMARY KEY, payload jsonb):

import json
from contextlib import closing
from datetime import timedelta
import pendulum
from airflow.sdk import dag, task
from airflow.providers.http.hooks.http import HttpHook
from airflow.providers.postgres.hooks.postgres import PostgresHook

@dag(schedule=None, start_date=pendulum.datetime(2026, 1, 1, tz="UTC"),
     catchup=False, max_active_runs=1)
def load_rest_api():
    @task(retries=2, retry_delay=timedelta(minutes=1))
    def load():
        response = HttpHook(method="GET", http_conn_id="source_api").run(
            endpoint="items", extra_options={"timeout": 30})
        response.raise_for_status()
        items = response.json()
        if not isinstance(items, list):
            raise ValueError("Ожидался массив объектов")
        rows = [(str(item["id"]), json.dumps(item)) for item in items]
        with closing(PostgresHook(postgres_conn_id="target_db").get_conn()) as conn:
            with conn.cursor() as cursor:
                cursor.executemany(
                    "INSERT INTO api_items(id, payload) VALUES (%s, %s::jsonb) "
                    "ON CONFLICT (id) DO UPDATE SET payload = EXCLUDED.payload",
                    rows,
                )
            conn.commit()
        return len(rows)
    load()

load_rest_api()

Connections задают адреса и доступы. Пример загружает один ответ целиком; для большого API добавьте пагинацию, строгую схему и ограниченные пакеты. В XCom возвращается только количество записей. В Airflow 2 импорт декораторов отличается.

Источник: Документация Apache Kafka.