Make `ExternalTaskSensor` work with Task SDK #48651

kaxil · 2025-04-02T00:23:37Z

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

gopidesupavan · 2025-04-02T17:49:52Z

Oh thanks kaxil , wrote similar way for the _get_count() for triggers as it also uses inside poke, i will fork this branch and will raise my changes in different PR for WorkflowTrigger.

kaxil · 2025-04-03T00:44:49Z

@gopidesupavan I had to add /execution/dag-runs/count endpoint too since the ExternalTaskSensor needed it

task-sdk/tests/task_sdk/api/test_client.py

gopidesupavan · 2025-04-03T04:51:34Z

@gopidesupavan I had to add /execution/dag-runs/count endpoint too since the ExternalTaskSensor needed it

thanks, will rebase mine, once it merged.

closes apache#47447

amoghrajesh

Few comments from initial look

amoghrajesh · 2025-04-03T08:25:04Z

airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py

+router = VersionedAPIRouter()
+
+ti_id_router = VersionedAPIRouter(


Can we add a comment here explaining why we need this one? It will be clear to reader too

amoghrajesh · 2025-04-03T08:27:11Z

airflow-core/src/airflow/api_fastapi/execution_api/routes/dag_runs.py

+@router.get("/count", status_code=status.HTTP_200_OK)
+def get_dr_count(
+    dag_id: str,
+    session: SessionDep,
+    logical_dates: Annotated[list[UtcDateTime] | None, Query()] = None,
+    run_ids: Annotated[list[str] | None, Query()] = None,
+    states: Annotated[list[str] | None, Query()] = None,
+) -> int:
+    """Get the count of DAG runs matching the given criteria."""
+    query = select(func.count()).select_from(DagRun).where(DagRun.dag_id == dag_id)
+
+    if logical_dates:
+        query = query.where(DagRun.logical_date.in_(logical_dates))
+
+    if run_ids:
+        query = query.where(DagRun.run_id.in_(run_ids))
+
+    if states:
+        query = query.where(DagRun.state.in_(states))
+
+    count = session.scalar(query)
+    return count or 0


We will need to add cadwyn migration for the new endpoints: https://docs.cadwyn.dev/concepts/version_changes/

I don't think so. We only need to add a migration for breaking changes (or changes to existing endpoints) from what I understand.

Just checked it here: https://docs.cadwyn.dev/concepts/endpoint_migrations/#defining-endpoints-that-didnt-exist-in-old-versions. Seems we will need it. Let me take it up

I just had a chat with the Cadwyn Author ( @zmievsa ) who recommends to only add it for breaking changes.

Depending on your needs. My general recommendation is to only add migrations for breaking changes
https://docs.cadwyn.dev/how_to/change_endpoints/#add-a-new-endpoint

I'll make Cadwyn's docs more verbose on when it makes the most sense to add a migration. Concepts section mostly focuses on what's possible with Cadwyn while "how to" focuses on what you should actually do.

Either way 99% of the time it makes sense to add an endpoint to all versions since it's not a breaking change. Your users will thank you later

Update: https://docs.cadwyn.dev/concepts/endpoint_migrations/#defining-endpoints-that-didnt-exist-in-old-versions added a bunch of notes here and there about this.

amoghrajesh · 2025-04-03T08:27:11Z

airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py


-@router.only_exists_in_older_versions
-@router.post(
+@router.get("/count", status_code=status.HTTP_200_OK)


We will need to add cadwyn migration for the new endpoints: https://docs.cadwyn.dev/concepts/version_changes/

I don't think so. We only need to add a migration for breaking changes (or changes to existing endpoints) from what I understand.

Same as #48651 (comment)

Check #48651 (comment) :)

airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_dag_runs.py

amoghrajesh · 2025-04-03T08:30:02Z

airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py

        assert response.json() == "2024-01-02T00:00:00Z"
+
+
+class TestGetCount:


Suggested change

class TestGetCount:

class TestGetTICount:

providers/standard/tests/unit/standard/sensors/test_external_task_sensor.py

amoghrajesh · 2025-04-03T08:37:59Z

providers/standard/tests/unit/standard/sensors/test_external_task_sensor.py

+    if AIRFLOW_V_3_0_PLUS:
+        dag_bag.bag_dag(dag=dag)
+    else:
+        dag_bag.bag_dag(dag=dag, root_dag=dag)



Lets just define a mini utility function, too many usages here

amoghrajesh · 2025-04-03T08:39:17Z

task-sdk/src/airflow/sdk/api/client.py

+        resp = self.client.get("task-instances/count", params=params)
+        return TICount(count=resp.json())


Safeguard it inside a try/except in case of a 500?

amoghrajesh · 2025-04-03T08:39:36Z

task-sdk/src/airflow/sdk/api/client.py

+        resp = self.client.get("dag-runs/count", params=params)
+        return DRCount(count=resp.json())


Safeguard in a try/except for cases like 500?

closes apache#47447 closes apache#47948

boring-cyborg bot added area:API Airflow's REST/HTTP API area:providers area:task-sdk provider:standard labels Apr 2, 2025

kaxil force-pushed the external-task-sensor branch from c2e5875 to b57ab8c Compare April 2, 2025 16:11

kaxil force-pushed the external-task-sensor branch 2 times, most recently from d272433 to 7566433 Compare April 3, 2025 00:42

kaxil marked this pull request as ready for review April 3, 2025 00:43

kaxil requested review from amoghrajesh and ashb as code owners April 3, 2025 00:43

kaxil commented Apr 3, 2025

View reviewed changes

kaxil force-pushed the external-task-sensor branch from 7566433 to 25d6863 Compare April 3, 2025 01:10

gopidesupavan approved these changes Apr 3, 2025

View reviewed changes

Make ExternalTaskSensor work with Task SDK

2526882

closes apache#47447

kaxil force-pushed the external-task-sensor branch from 25d6863 to 2526882 Compare April 3, 2025 07:41

kaxil merged commit 6775bf7 into apache:main Apr 3, 2025
90 checks passed

kaxil deleted the external-task-sensor branch April 3, 2025 08:27

amoghrajesh reviewed Apr 3, 2025

View reviewed changes

nailo2c pushed a commit to nailo2c/airflow that referenced this pull request Apr 4, 2025

Make ExternalTaskSensor work with Task SDK (apache#48651)

521639e

closes apache#47447 closes apache#47948

eladkal mentioned this pull request Apr 6, 2025

Status of testing Providers that were prepared on April 06, 2025 #48842

Closed

diogotrodrigues pushed a commit to diogotrodrigues/airflow that referenced this pull request Apr 6, 2025

Make ExternalTaskSensor work with Task SDK (apache#48651)

3b2cae9

closes apache#47447 closes apache#47948

simonprydden pushed a commit to simonprydden/airflow that referenced this pull request Apr 8, 2025

Make ExternalTaskSensor work with Task SDK (apache#48651)

21f9bde

closes apache#47447 closes apache#47948

		router = VersionedAPIRouter()

		ti_id_router = VersionedAPIRouter(

		assert response.json() == "2024-01-02T00:00:00Z"


		class TestGetCount:

		resp = self.client.get("task-instances/count", params=params)
		return TICount(count=resp.json())

		resp = self.client.get("dag-runs/count", params=params)
		return DRCount(count=resp.json())

Make ExternalTaskSensor work with Task SDK #48651

Make ExternalTaskSensor work with Task SDK #48651

Uh oh!

Conversation

kaxil commented Apr 2, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

gopidesupavan commented Apr 2, 2025

Uh oh!

kaxil commented Apr 3, 2025

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

gopidesupavan commented Apr 3, 2025

Uh oh!

Uh oh!

amoghrajesh left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zmievsa Apr 5, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

Make `ExternalTaskSensor` work with Task SDK #48651

Make `ExternalTaskSensor` work with Task SDK #48651

kaxil commented Apr 2, 2025 •

edited

Loading

zmievsa Apr 5, 2025 •

edited

Loading