From dc63b6aa7706eaee3cb45fbc24f3bd231a99eb05 Mon Sep 17 00:00:00 2001 From: Alejandro Morera Date: Tue, 18 Aug 2026 16:30:29 -0600 Subject: [PATCH 1/4] Add DAG parse times via REST API how-to --- docs/how-tos/my_airflow/dag-parse-times.md | 93 ++++++++++++++++++++++ 1 file changed, 93 insertions(+) create mode 100644 docs/how-tos/my_airflow/dag-parse-times.md diff --git a/docs/how-tos/my_airflow/dag-parse-times.md b/docs/how-tos/my_airflow/dag-parse-times.md new file mode 100644 index 0000000..5ffab7a --- /dev/null +++ b/docs/how-tos/my_airflow/dag-parse-times.md @@ -0,0 +1,93 @@ +--- +title: "Find Slow-Parsing DAGs with the REST API" +sidebar_label: "DAG parsing times" +description: "Use the Airflow REST API to list DAG parse durations and find slow-parsing DAGs in Team Airflow and My Airflow on Datacoves." +sidebar_position: 79 +--- +# Find slow-parsing DAGs with the REST API + +Every DAG file is re-parsed continuously by the dag-processor. A DAG that is slow to parse (top-level code doing imports, API calls, or file reads) delays scheduling for every DAG in the environment, so parse times are one of the first things to check when Airflow feels sluggish. + +On Airflow 3 environments, the dag-processor records each DAG's most recent parse duration and the REST API exposes it, so you can get the real parsing-time report with a simple API call: no access to the dag-processor pod is needed. + +:::info +This works on Airflow 3 environments, for both **Team Airflow** and **My Airflow** (it is the same REST API). On My Airflow you can also use the [`datacoves my parse-logs`](/docs/reference/airflow/datacoves-commands#datacoves-my-parse-logs) command, which additionally shows the parse log of individual DAG files. + +Airflow 2 does not expose parse durations through its API; see [the note below](#airflow-2-environments). +::: + +## What you need + +1. An API key and the API URL for your target instance: follow [How to use the My Airflow API](/docs/how-tos/my_airflow/use-my-airflow-api). The same key works against Team Airflow if you have access; for Team Airflow, the API base is your Airflow URL (the one in your browser) plus `/api/v2`. +2. Store both in a `.env` file (and add it to `.gitignore`): + +```env +AIRFLOW_API_URL = "https://your-airflow-url/api/v2" +AIRFLOW_API_KEY = "your-api-key-here" +``` + +## Get the parsing-time report + +The `GET /dags` endpoint returns `last_parse_duration` (seconds) and `last_parsed_time` for every DAG. The script below fetches all DAGs (paginated) and prints them slowest-first, the same shape as the `datacoves my parse-logs` report: + +```python +# dag_parse_times.py +import os + +import requests +from dotenv import load_dotenv + +load_dotenv() +API_URL = os.getenv("AIRFLOW_API_URL").rstrip("/") +API_KEY = os.getenv("AIRFLOW_API_KEY") +HEADERS = {"Authorization": f"Token {API_KEY}"} + + +def fetch_all_dags(): + dags, offset = [], 0 + while True: + response = requests.get( + f"{API_URL}/dags", + headers=HEADERS, + params={"limit": 100, "offset": offset}, + ) + response.raise_for_status() + payload = response.json() + dags.extend(payload["dags"]) + offset += 100 + if offset >= payload["total_entries"]: + return dags + + +def print_parse_report(dags): + dags.sort(key=lambda d: d.get("last_parse_duration") or 0, reverse=True) + print(f"{'DAG':<50} {'Parse time':>12} {'Last parsed'}") + for dag in dags: + duration = dag.get("last_parse_duration") + duration = f"{duration:.3f}s" if duration is not None else "-" + parsed_at = dag.get("last_parsed_time") or "-" + print(f"{dag['dag_id']:<50} {duration:>12} {parsed_at}") + + +if __name__ == "__main__": + print_parse_report(fetch_all_dags()) +``` + +Example output: + +``` +DAG Parse time Last parsed +daily_loan_run 2.153s 2026-08-18T14:02:11.480217Z +variables_python_script 0.310s 2026-08-18T14:02:10.912304Z +simple_bash_dag 0.052s 2026-08-18T14:02:10.598721Z +``` + +A parse time consistently above ~1 second is worth investigating: the usual cause is work done at the top level of the DAG file (imports of heavy libraries, `Variable.get` outside a task, reading files or calling APIs during parsing). See [Airflow's best practices on top-level code](https://airflow.apache.org/docs/apache-airflow/stable/best-practices.html#top-level-python-code) for how to restructure it. + +:::tip +`last_parse_duration` reflects the most recent parse by the environment's dag-processor -- the number that actually matters for scheduling latency. Running `airflow dags report` inside a DAG task measures a worker's copy of the files under different conditions and can be misleading. +::: + +## Airflow 2 environments + +Airflow 2 records when each DAG was last parsed (`last_parsed_time`) but not how long it took, and its REST API does not expose parse durations. This is a known limitation; parse times there only exist in the dag-processor manager logs. Since new Datacoves environments run Airflow 3, no Airflow 2-specific tooling is provided -- upgrading is the path to this report. From 45c5fa2c3eb0dd98fbb9c4f229c7a9fb5ef92abb Mon Sep 17 00:00:00 2001 From: Alejandro Morera Date: Tue, 18 Aug 2026 16:52:57 -0600 Subject: [PATCH 2/4] Use the Airflow 3 JWT auth flow in the parse times guide --- docs/how-tos/my_airflow/dag-parse-times.md | 39 ++++++++++++++-------- 1 file changed, 26 insertions(+), 13 deletions(-) diff --git a/docs/how-tos/my_airflow/dag-parse-times.md b/docs/how-tos/my_airflow/dag-parse-times.md index 5ffab7a..d8071cb 100644 --- a/docs/how-tos/my_airflow/dag-parse-times.md +++ b/docs/how-tos/my_airflow/dag-parse-times.md @@ -11,24 +11,27 @@ Every DAG file is re-parsed continuously by the dag-processor. A DAG that is slo On Airflow 3 environments, the dag-processor records each DAG's most recent parse duration and the REST API exposes it, so you can get the real parsing-time report with a simple API call: no access to the dag-processor pod is needed. :::info -This works on Airflow 3 environments, for both **Team Airflow** and **My Airflow** (it is the same REST API). On My Airflow you can also use the [`datacoves my parse-logs`](/docs/reference/airflow/datacoves-commands#datacoves-my-parse-logs) command, which additionally shows the parse log of individual DAG files. +This guide targets **Team Airflow** on Airflow 3 environments. For **My Airflow**, use the [`datacoves my parse-logs`](/docs/reference/airflow/datacoves-commands#datacoves-my-parse-logs) command instead, which shows the same report plus the parse log of individual DAG files. Airflow 2 does not expose parse durations through its API; see [the note below](#airflow-2-environments). ::: ## What you need -1. An API key and the API URL for your target instance: follow [How to use the My Airflow API](/docs/how-tos/my_airflow/use-my-airflow-api). The same key works against Team Airflow if you have access; for Team Airflow, the API base is your Airflow URL (the one in your browser) plus `/api/v2`. -2. Store both in a `.env` file (and add it to `.gitignore`): +1. A Datacoves API key: generate one following [How to use the My Airflow API](/docs/how-tos/my_airflow/use-my-airflow-api). The same key authenticates you against the Team Airflow of **any environment you have access to**, with your own permissions. +2. The Airflow URL of the target environment: the same one you open in your browser (`https://airflow-.`). +3. Store both in a `.env` file (and add it to `.gitignore`): ```env -AIRFLOW_API_URL = "https://your-airflow-url/api/v2" -AIRFLOW_API_KEY = "your-api-key-here" +AIRFLOW_URL = "https://airflow-your-env.your-domain.com" +DATACOVES_API_KEY = "your-api-key-here" ``` +The script works the same from your laptop, from CI, or from a Datacoves workbench terminal -- the Airflow API endpoints are reachable from all of them. + ## Get the parsing-time report -The `GET /dags` endpoint returns `last_parse_duration` (seconds) and `last_parsed_time` for every DAG. The script below fetches all DAGs (paginated) and prints them slowest-first, the same shape as the `datacoves my parse-logs` report: +Airflow 3's API uses short-lived JWTs: you first exchange your Datacoves API key for a JWT at `POST /auth/token` (using the special `__datacoves_token__` username), then call the API with it. The `GET /api/v2/dags` endpoint returns `last_parse_duration` (seconds) and `last_parsed_time` for every DAG. The script below fetches all DAGs (paginated) and prints them slowest-first, the same shape as the `datacoves my parse-logs` report: ```python # dag_parse_times.py @@ -38,17 +41,26 @@ import requests from dotenv import load_dotenv load_dotenv() -API_URL = os.getenv("AIRFLOW_API_URL").rstrip("/") -API_KEY = os.getenv("AIRFLOW_API_KEY") -HEADERS = {"Authorization": f"Token {API_KEY}"} +BASE_URL = os.getenv("AIRFLOW_URL").rstrip("/") +API_KEY = os.getenv("DATACOVES_API_KEY") + + +def get_jwt(): + """Exchange the Datacoves API key for a short-lived Airflow JWT.""" + response = requests.post( + f"{BASE_URL}/auth/token", + json={"username": "__datacoves_token__", "password": API_KEY}, + ) + response.raise_for_status() + return response.json()["access_token"] -def fetch_all_dags(): +def fetch_all_dags(headers): dags, offset = [], 0 while True: response = requests.get( - f"{API_URL}/dags", - headers=HEADERS, + f"{BASE_URL}/api/v2/dags", + headers=headers, params={"limit": 100, "offset": offset}, ) response.raise_for_status() @@ -70,7 +82,8 @@ def print_parse_report(dags): if __name__ == "__main__": - print_parse_report(fetch_all_dags()) + headers = {"Authorization": f"Bearer {get_jwt()}"} + print_parse_report(fetch_all_dags(headers)) ``` Example output: From 93ef125014284232987954d3e29e89eca78bc4b7 Mon Sep 17 00:00:00 2001 From: Alejandro Morera Date: Wed, 19 Aug 2026 08:34:36 -0600 Subject: [PATCH 3/4] Move the DAG parse times guide under Team Airflow how-tos --- docs/how-tos/{my_airflow => airflow}/dag-parse-times.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) rename docs/how-tos/{my_airflow => airflow}/dag-parse-times.md (99%) diff --git a/docs/how-tos/my_airflow/dag-parse-times.md b/docs/how-tos/airflow/dag-parse-times.md similarity index 99% rename from docs/how-tos/my_airflow/dag-parse-times.md rename to docs/how-tos/airflow/dag-parse-times.md index d8071cb..3a6a20c 100644 --- a/docs/how-tos/my_airflow/dag-parse-times.md +++ b/docs/how-tos/airflow/dag-parse-times.md @@ -2,7 +2,7 @@ title: "Find Slow-Parsing DAGs with the REST API" sidebar_label: "DAG parsing times" description: "Use the Airflow REST API to list DAG parse durations and find slow-parsing DAGs in Team Airflow and My Airflow on Datacoves." -sidebar_position: 79 +sidebar_position: 40 --- # Find slow-parsing DAGs with the REST API From f9795e87c7c24dc6f7c0aadecfb3eeadb56a040f Mon Sep 17 00:00:00 2001 From: Alejandro Morera Date: Wed, 19 Aug 2026 08:46:21 -0600 Subject: [PATCH 4/4] Use the environment Airflow API token as the primary auth path --- docs/how-tos/airflow/dag-parse-times.md | 50 ++++++++++++++----------- 1 file changed, 28 insertions(+), 22 deletions(-) diff --git a/docs/how-tos/airflow/dag-parse-times.md b/docs/how-tos/airflow/dag-parse-times.md index 3a6a20c..76172b2 100644 --- a/docs/how-tos/airflow/dag-parse-times.md +++ b/docs/how-tos/airflow/dag-parse-times.md @@ -18,20 +18,20 @@ Airflow 2 does not expose parse durations through its API; see [the note below]( ## What you need -1. A Datacoves API key: generate one following [How to use the My Airflow API](/docs/how-tos/my_airflow/use-my-airflow-api). The same key authenticates you against the Team Airflow of **any environment you have access to**, with your own permissions. -2. The Airflow URL of the target environment: the same one you open in your browser (`https://airflow-.`). +1. An Airflow API token for the target environment: in the launchpad, go to **Environments**, open your environment's **Keys**, and click **Generate New Airflow Token**. Copy the token immediately -- it is only shown once. +2. The **Airflow API URL** shown on that same page (`https://airflow-./api/v2/`). 3. Store both in a `.env` file (and add it to `.gitignore`): ```env -AIRFLOW_URL = "https://airflow-your-env.your-domain.com" -DATACOVES_API_KEY = "your-api-key-here" +AIRFLOW_API_URL = "https://airflow-your-env.your-domain.com/api/v2/" +AIRFLOW_API_TOKEN = "your-token-here" ``` The script works the same from your laptop, from CI, or from a Datacoves workbench terminal -- the Airflow API endpoints are reachable from all of them. ## Get the parsing-time report -Airflow 3's API uses short-lived JWTs: you first exchange your Datacoves API key for a JWT at `POST /auth/token` (using the special `__datacoves_token__` username), then call the API with it. The `GET /api/v2/dags` endpoint returns `last_parse_duration` (seconds) and `last_parsed_time` for every DAG. The script below fetches all DAGs (paginated) and prints them slowest-first, the same shape as the `datacoves my parse-logs` report: +The `GET /dags` endpoint returns `last_parse_duration` (seconds) and `last_parsed_time` for every DAG. The script below fetches all DAGs (paginated) and prints them slowest-first, the same shape as the `datacoves my parse-logs` report: ```python # dag_parse_times.py @@ -41,26 +41,16 @@ import requests from dotenv import load_dotenv load_dotenv() -BASE_URL = os.getenv("AIRFLOW_URL").rstrip("/") -API_KEY = os.getenv("DATACOVES_API_KEY") +API_URL = os.getenv("AIRFLOW_API_URL").rstrip("/") +HEADERS = {"Authorization": f"Bearer {os.getenv('AIRFLOW_API_TOKEN')}"} -def get_jwt(): - """Exchange the Datacoves API key for a short-lived Airflow JWT.""" - response = requests.post( - f"{BASE_URL}/auth/token", - json={"username": "__datacoves_token__", "password": API_KEY}, - ) - response.raise_for_status() - return response.json()["access_token"] - - -def fetch_all_dags(headers): +def fetch_all_dags(): dags, offset = [], 0 while True: response = requests.get( - f"{BASE_URL}/api/v2/dags", - headers=headers, + f"{API_URL}/dags", + headers=HEADERS, params={"limit": 100, "offset": offset}, ) response.raise_for_status() @@ -82,8 +72,7 @@ def print_parse_report(dags): if __name__ == "__main__": - headers = {"Authorization": f"Bearer {get_jwt()}"} - print_parse_report(fetch_all_dags(headers)) + print_parse_report(fetch_all_dags()) ``` Example output: @@ -101,6 +90,23 @@ A parse time consistently above ~1 second is worth investigating: the usual caus `last_parse_duration` reflects the most recent parse by the environment's dag-processor -- the number that actually matters for scheduling latency. Running `airflow dags report` inside a DAG task measures a worker's copy of the files under different conditions and can be misleading. ::: +## Alternative: authenticate as your own user + +The Airflow API token above belongs to the environment's service account and is a JWT with an expiration date, so long-lived automation would need re-generating it. As an alternative, you can authenticate with your personal Datacoves API key (generated under [User Settings > My Airflow API](/docs/how-tos/my_airflow/use-my-airflow-api) -- despite the name, it is a general Datacoves key): exchange it for a short-lived Airflow JWT at `POST /auth/token` using the special `__datacoves_token__` username. This authenticates you as **your own user** with your permissions, the key does not expire, and the same key works against **any environment you have access to**: + +```python +def get_jwt(base_url, datacoves_api_key): + """Exchange a Datacoves API key for a short-lived Airflow JWT.""" + response = requests.post( + f"{base_url}/auth/token", + json={"username": "__datacoves_token__", "password": datacoves_api_key}, + ) + response.raise_for_status() + return response.json()["access_token"] +``` + +Use the returned token in the same `Authorization: Bearer` header as above (`base_url` is the Airflow URL without `/api/v2`). + ## Airflow 2 environments Airflow 2 records when each DAG was last parsed (`last_parsed_time`) but not how long it took, and its REST API does not expose parse durations. This is a known limitation; parse times there only exist in the dag-processor manager logs. Since new Datacoves environments run Airflow 3, no Airflow 2-specific tooling is provided -- upgrading is the path to this report.