Schedule Trends API from an Airflow DAG

An Airflow DAG is a scheduled HTTP client. POST https://api.trendsapi.ai/api with a Bearer key. Parse the string body before XCom or a load. This is not crontab. Retries, pools, and downstream warehouse tasks are the reason to use a DAG. A single machine timer is on cron. Fields: the API reference. Caps: pricing.

PythonOperator posts, then parses

import json, os, requests
from airflow.decorators import task

@task(retries=3, retry_delay=300)
def growth(keyword: str):
    resp = requests.post(
        "https://api.trendsapi.ai/api",
        headers={
            "Authorization": f"Bearer {os.environ['TRENDSAPI_API_KEY']}",
            "Content-Type": "application/json",
        },
        json={
            "mode": "get_growth",
            "source": "google search",
            "keyword": keyword,
            "percent_growth": ["3M", "12M"],
        },
        timeout=60,
    )
    resp.raise_for_status()
    return json.loads(resp.json()["body"])

HttpOperator from the HTTP provider can send the same POST. The next task still needs json.loads on body. Do not XCom the envelope string. The parse rule is on parse the body.

Put the key in a Variable or a secret backend. The DAG file is not a vault.

One get_growth, not five series tasks

Five windows in percent_growth are one successful POST. Five get_time_series tasks are five. Official usage notes count each 200. Collapse windows on get_growth. Map over keywords, not over presets.

On 429, the task fails and retries. Tight-looping inside the callable burns the same cap. Error codes sit on errors.

After the parse, write out

A small results list can go to BigQuery through a load task. A weekly series belongs in object storage or a table with date, value, keyword, source. Warehouse shape is on BigQuery. Airflow schedules the POST. It does not replace a warehouse schema.