You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Frost3281 (via GitHub)" <gi...@apache.org> on 2025/04/30 05:19:11 UTC
[D] ExternalTaskSensor (only for last dag runs) in Airflow 3 [airflow]
GitHub user Frost3281 created a discussion: ExternalTaskSensor (only for last dag runs) in Airflow 3
In Airflow 2, I used the following Sensor to check the status of the latest run of external DAGs. In Airflow 3, this code no longer works.
"`Direct database access via the ORM is not allowed in Airflow 3.0`"
I have tried many options: Engine / Connection and other methods of database access. Is there a proper way in Airflow 3 to achieve the same functionality with this operator?
```python
import json
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, cast
import pytz # type: ignore
from airflow.exceptions import TaskNotFound
from airflow.models import DagBag, DagModel
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.utils.db import provide_session # type: ignore
from sqlalchemy import select
from sqlalchemy.orm import Session
DAGS_TO_EXTERNAL_TASK_DOCS: dict[str, str] = {}
class LastDagRunSensor(ExternalTaskSensor):
"""Sensor to retrieve the last run of a DAG."""
def __init__(
self,
external_dag_id: str,
external_task_id: str | None = None,
**kwargs: Any,
) -> None:
"""Initializer."""
def dag_last_exec(crnt_dttm: datetime) -> datetime: # noqa: ARG001
return self.get_dag_last_execution_date(self.external_dag_id) # type: ignore
super().__init__(
external_dag_id=external_dag_id,
external_task_id=external_task_id,
**kwargs,
)
self.execution_date_fn = dag_last_exec
@provide_session
def get_dag_last_execution_date(
self,
dag_id: str, # noqa: ARG002
session: Session,
) -> datetime:
"""Get the last execution date of the DAG."""
dag = _get_dag_by_id(session, self.external_dag_id)
if not dag:
return _yesterday()
return dag.get_last_dagrun(include_externally_triggered=True).execution_date
def _get_dag_by_id(session: Session, dag_id: str) -> DagModel | None:
query = select(DagModel).where(DagModel.dag_id == dag_id)
return session.execute(query).scalars().first()
def _yesterday() -> datetime:
return datetime.now(tz=pytz.timezone('Europe/Moscow')) - timedelta(days=1)
GitHub link: https://github.com/apache/airflow/discussions/49997
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] ExternalTaskSensor (only for last dag runs) in Airflow 3 [airflow]
Posted by "ajbosco (via GitHub)" <gi...@apache.org>.
GitHub user ajbosco added a comment to the discussion: ExternalTaskSensor (only for last dag runs) in Airflow 3
Hi @potiuk, the use case is that I simply want to be able to take a DAG ID, a Task ID, and a range of dates then determine if that DAG ID/Task ID has completed within that time. This was trivial in Airflow 2.0 and is really the only way an External Task Sensor makes any sense.
GitHub link: https://github.com/apache/airflow/discussions/49997#discussioncomment-14021397
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] ExternalTaskSensor (only for last dag runs) in Airflow 3 [airflow]
Posted by "ajbosco (via GitHub)" <gi...@apache.org>.
GitHub user ajbosco edited a comment on the discussion: ExternalTaskSensor (only for last dag runs) in Airflow 3
I tried `get_ti_count`, but it takes a list of `logical_dates` not a start/end date.
GitHub link: https://github.com/apache/airflow/discussions/49997#discussioncomment-14036653
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] ExternalTaskSensor (only for last dag runs) in Airflow 3 [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
GitHub user potiuk added a comment to the discussion: ExternalTaskSensor (only for last dag runs) in Airflow 3
> I have tried many options: Engine / Connection and other methods of database access. And I tried access through API (always getting connection refused). Is there a proper way in Airflow 3 to achieve the same functionality with this operator?
Nope. But if you get "connection refused" it likely means you are using wrong hostname, wrong port, or you have to configure your networking to allow your workers to connect to your api_server (using the right host and port).
Connection refused always means tha when you try to connect to a give host and port, there is nothing on the other side to listen to your connection - so It's either firewall/networking blocking you, or you use wrong host/port.
GitHub link: https://github.com/apache/airflow/discussions/49997#discussioncomment-13835032
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] ExternalTaskSensor (only for last dag runs) in Airflow 3 [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
GitHub user potiuk added a comment to the discussion: ExternalTaskSensor (only for last dag runs) in Airflow 3
> As far as I can tell, the API does not expose the ability to do something like this:
@ajbosco - I am sure you can run appropriate python code to query REST API - did you actually try it?
And if you think there is a use case which might be also useful for others - to be "easily" accessible via Task SDK API - as explained before - please make a feature request, describing the use case and why you think it's good to have it as a "feature" in the Task SDKL - it's all in your hands to describe it well enough and justify that this case is useful for many users. Because we are optimising for many users. It's a little sad to hear but if there is a case that useful for you only- maintainers who give you the software for free - often working in their evenings and weekends will not optimise the acess for "you" only, That would be quite selfish. But if you justify that it is useful for others as well, and nicely describe it in a request, there are already things we added betwen 3.0.0 and 3.0.4 that we got convinced are useful for many users.
But it's up to you to describe it in the way that will be convincing. All in your hands.
GitHub link: https://github.com/apache/airflow/discussions/49997#discussioncomment-14038689
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] ExternalTaskSensor (only for last dag runs) in Airflow 3 [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
GitHub user potiuk added a comment to the discussion: ExternalTaskSensor (only for last dag runs) in Airflow 3
I guess https://github.com/apache/airflow/pull/53655 scheduled for 3.0.4 might address it. Please check if it does. If it does not - please open a new issue explaining your use case and what you want to achieve. Similarly as https://github.com/apache/airflow/pull/53655 when we see that particular way of accessing informaiton from Airflow is useful by many users and necessary to be done without REST API we can consider adding it to task SDK.
The key is that you clearly explain the use case in an issue and justify why you need it (providing that the upcoming changes do not already solve your issue).
GitHub link: https://github.com/apache/airflow/discussions/49997#discussioncomment-13868060
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] ExternalTaskSensor (only for last dag runs) in Airflow 3 [airflow]
Posted by "ajbosco (via GitHub)" <gi...@apache.org>.
GitHub user ajbosco edited a comment on the discussion: ExternalTaskSensor (only for last dag runs) in Airflow 3
As far as I can tell, the API does not expose the ability to do something like this:
```
if self.lookback_date == "run_date":
count = (
session.query(TI)
.filter(
TI.dag_id == self.external_dag_id,
TI.task_id == self.external_task_id,
TI.state.in_(self.allowed_states),
TI.start_date >= start_dttm,
TI.start_date <= end_dttm,
)
.count()
)
else:
count = (
session.query(TI)
.filter(
TI.dag_id == self.external_dag_id,
TI.task_id == self.external_task_id,
TI.state.in_(self.allowed_states),
TI.logical_date >= start_dttm,
TI.logical_date <= end_dttm,
)
.count()
)
GitHub link: https://github.com/apache/airflow/discussions/49997#discussioncomment-14036535
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] ExternalTaskSensor (only for last dag runs) in Airflow 3 [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
GitHub user gopidesupavan added a comment to the discussion: ExternalTaskSensor (only for last dag runs) in Airflow 3
https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html#operation/get_task_instances_batch
https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html#operation/get_dag_runs
With the above combination of endpoints you could workout easily IMHO.
Yes agree with @potiuk if you clear requirement use case and why its needed support in API , please raise issue and explain
GitHub link: https://github.com/apache/airflow/discussions/49997#discussioncomment-14045389
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] ExternalTaskSensor (only for last dag runs) in Airflow 3 [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
GitHub user potiuk edited a comment on the discussion: ExternalTaskSensor (only for last dag runs) in Airflow 3
> As far as I can tell, the API does not expose the ability to do something like this:
@ajbosco - I am sure you can run appropriate python code to query REST API - did you actually try it?
And if you think there is a use case which might be also useful for others - to be "easily" accessible via Task SDK API - as explained before - please make a feature request, describing the use case and why you think it's good to have it as a "feature" in the Task SDKL - it's all in your hands to describe it well enough and justify that this case is useful for many users. Because we are optimising for many users. It's a little sad to hear but if there is a case that useful for you only- maintainers who give you the software for free - often working in their evenings and weekends will not optimise the acess for "you" only, That would be quite selfish from your side if you think it's justified to make it easy only for you.
But if you justify that it is useful for others as well, and nicely describe it in a request, there are already things we added betwen 3.0.0 and 3.0.4 that we got convinced are useful for many users and added.
But it's up to you to describe it in the way that will be convincing. All in your hands.
GitHub link: https://github.com/apache/airflow/discussions/49997#discussioncomment-14038689
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] ExternalTaskSensor (only for last dag runs) in Airflow 3 [airflow]
Posted by "Frost3281 (via GitHub)" <gi...@apache.org>.
GitHub user Frost3281 edited a discussion: ExternalTaskSensor (only for last dag runs) in Airflow 3
In Airflow 2, I used the following Sensor to check the status of the latest run of external DAGs. In Airflow 3, this code no longer works.
"`Direct database access via the ORM is not allowed in Airflow 3.0`"
I have tried many options: Engine / Connection and other methods of database access. And I tried access through API (always getting connection refused). Is there a proper way in Airflow 3 to achieve the same functionality with this operator?
```python
import json
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, cast
import pytz # type: ignore
from airflow.exceptions import TaskNotFound
from airflow.models import DagBag, DagModel
from airflow.sensors.external_task import ExternalTaskSensor
from airflow.utils.db import provide_session # type: ignore
from sqlalchemy import select
from sqlalchemy.orm import Session
DAGS_TO_EXTERNAL_TASK_DOCS: dict[str, str] = {}
class LastDagRunSensor(ExternalTaskSensor):
"""Sensor to retrieve the last run of a DAG."""
def __init__(
self,
external_dag_id: str,
external_task_id: str | None = None,
**kwargs: Any,
) -> None:
"""Initializer."""
def dag_last_exec(crnt_dttm: datetime) -> datetime: # noqa: ARG001
return self.get_dag_last_execution_date(self.external_dag_id) # type: ignore
super().__init__(
external_dag_id=external_dag_id,
external_task_id=external_task_id,
**kwargs,
)
self.execution_date_fn = dag_last_exec
@provide_session
def get_dag_last_execution_date(
self,
dag_id: str, # noqa: ARG002
session: Session,
) -> datetime:
"""Get the last execution date of the DAG."""
dag = _get_dag_by_id(session, self.external_dag_id)
if not dag:
return _yesterday()
return dag.get_last_dagrun(include_externally_triggered=True).execution_date
def _get_dag_by_id(session: Session, dag_id: str) -> DagModel | None:
query = select(DagModel).where(DagModel.dag_id == dag_id)
return session.execute(query).scalars().first()
def _yesterday() -> datetime:
return datetime.now(tz=pytz.timezone('Europe/Moscow')) - timedelta(days=1)
GitHub link: https://github.com/apache/airflow/discussions/49997
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] ExternalTaskSensor (only for last dag runs) in Airflow 3 [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
GitHub user gopidesupavan added a comment to the discussion: ExternalTaskSensor (only for last dag runs) in Airflow 3
There are already functionality implemeted for getting count and states etc; look at this
https://github.com/apache/airflow/blob/main/providers/standard/src/airflow/providers/standard/triggers/external_task.py#L142
```
from airflow.sdk.execution_time.task_runner import RuntimeTaskInstance
RuntimeTaskInstance.get_ti_count
RuntimeTaskInstance.get_task_states
RuntimeTaskInstance.get_dr_count
RuntimeTaskInstance.get_dagrun_state
RuntimeTaskInstance.get_dr_count
```
You can take a look and see whats useful in your scenario for above
GitHub link: https://github.com/apache/airflow/discussions/49997#discussioncomment-14036618
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] ExternalTaskSensor (only for last dag runs) in Airflow 3 [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
GitHub user gopidesupavan edited a comment on the discussion: ExternalTaskSensor (only for last dag runs) in Airflow 3
https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html#operation/get_task_instances_batch
https://airflow.apache.org/docs/apache-airflow/stable/stable-rest-api-ref.html#operation/get_dag_runs
With the above combination of endpoints you could workout easily IMHO.
Yes agree with @potiuk, if you have clear requirement use case and why its needed support in API , please raise issue and explain
GitHub link: https://github.com/apache/airflow/discussions/49997#discussioncomment-14045389
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] ExternalTaskSensor (only for last dag runs) in Airflow 3 [airflow]
Posted by "zerodarkzone (via GitHub)" <gi...@apache.org>.
GitHub user zerodarkzone edited a comment on the discussion: ExternalTaskSensor (only for last dag runs) in Airflow 3
Hi, I was doing something similar in airflow 2. This is a pretty big breaking change, sensors like 'ExternalTaskSensor' are pretty useless without being able to get the last execution date of a dag. Is there an easy way to solve this?
GitHub link: https://github.com/apache/airflow/discussions/49997#discussioncomment-13866121
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] ExternalTaskSensor (only for last dag runs) in Airflow 3 [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
GitHub user potiuk added a comment to the discussion: ExternalTaskSensor (only for last dag runs) in Airflow 3
Security trumps convenience, sorry. There is no way the fact that you can write few less line of code makes it an argument to be able to drop the whole database. The impact of those two things is incomparable.
And with REST API and python client of ours - you do not even have to make it very complex. Python client exposes Pythonic interface that you can use to query for stuff. Being a little sarcastic - I know it;s an awfully lot of work to install one more package, configure a user that can be used via REST API and run maybe 10 lines of code to do the same things with python client and REST API. Sorry for making you to do all that work, I know it's really difficult, but in exchange you get much more secure setup of Airflow.
You are welcome.
GitHub link: https://github.com/apache/airflow/discussions/49997#discussioncomment-14036367
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] ExternalTaskSensor (only for last dag runs) in Airflow 3 [airflow]
Posted by "ajbosco (via GitHub)" <gi...@apache.org>.
GitHub user ajbosco added a comment to the discussion: ExternalTaskSensor (only for last dag runs) in Airflow 3
As far as I can tell, the API does not expose the ability to do something like this:
```
if self.lookback_date == "run_date":
count = (
session.query(TI)
.filter(
TI.dag_id == self.external_dag_id,
TI.task_id == self.external_task_id,
TI.state.in_(self.allowed_states),
TI.start_date >= start_dttm,
TI.start_date <= end_dttm,
)
.count()
)
else:
count = (
session.query(TI)
.filter(
TI.dag_id == self.external_dag_id,
TI.task_id == self.external_task_id,
TI.state.in_(self.allowed_states),
TI.logical_date >= start_dttm,
TI.logical_date <= end_dttm,
)
.count()
)
```
GitHub link: https://github.com/apache/airflow/discussions/49997#discussioncomment-14036535
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] ExternalTaskSensor (only for last dag runs) in Airflow 3 [airflow]
Posted by "BioQwer (via GitHub)" <gi...@apache.org>.
GitHub user BioQwer added a comment to the discussion: ExternalTaskSensor (only for last dag runs) in Airflow 3
Hi everyone, I have a similar application and a similar problem.
I have many DAGs that run once a day but at different times.
We also queried the database to find out the start times in order to understand the delta for the External Task Sensor.
This problem is a blocker for us since we have around 100 DAGs with inter-dependencies. Figuring out the delta between each one is a massive undertaking that could introduce many errors.
Now I understand how to get this from the API.
`/api/v2/dags`
i can get diff from `next_dagrun_logical_date`
```
{
"dags": [
{
"dag_id": "test",
"dag_display_name": "test",
"is_paused": true,
"is_stale": false,
"last_parsed_time": "2026-06-05T05:50:35.210927Z",
"last_parse_duration": 0.5269125308841467,
"last_expired": null,
"bundle_name": "dags-folder",
"description": "",
"timetable_summary": "00 21 * * 0",
"timetable_description": "At 21:00, only on Sunday",
"timetable_partitioned": false,
"next_dagrun_logical_date": "2026-05-31T18:00:00Z",
"next_dagrun_data_interval_start": "2026-05-31T18:00:00Z",
"next_dagrun_data_interval_end": "2026-05-31T18:00:00Z",
"next_dagrun_run_after": "2026-05-31T18:00:00Z",
"allowed_run_types": null,
...
},
```
But I have absolutely no idea how to obtain a technical client for the requests.
The client needs a token, but if it is running inside the scheduler, it already has all the necessary data, tokens, and other things.
```python
from airflow.sdk.api.client import Client
class Client(httpx.Client):
def __init__(self, *, base_url: str | None, dry_run: bool = False, token: str, **kwargs: Any):
...
```
It turns out we will be querying the time inside the scheduler.
There is a Local client, but its API is limited.
<img width="761" height="370" alt="image" src="https://github.com/user-attachments/assets/163b0be3-62e6-4be7-a26a-af462fb758f0" />
Can you advise me on how to proceed correctly in this situation? Airflow version 3.2.2.
GitHub link: https://github.com/apache/airflow/discussions/49997#discussioncomment-17188652
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] ExternalTaskSensor (only for last dag runs) in Airflow 3 [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
GitHub user potiuk edited a comment on the discussion: ExternalTaskSensor (only for last dag runs) in Airflow 3
> As far as I can tell, the API does not expose the ability to do something like this:
@ajbosco - I am sure you can run appropriate python code to query REST API - did you actually try it?
And if you think there is a use case which might be also useful for others - to be "easily" accessible via Task SDK API - as explained before - please make a feature request, describing the use case and why you think it's good to have it as a "feature" in the Task SDKL - it's all in your hands to describe it well enough and justify that this case is useful for many users. Because we are optimising for many users. It's a little sad to hear but if there is a case that useful for you only- maintainers who give you the software for free - often working in their evenings and weekends will not optimise the acess for "you" only, That would be quite selfish from your side if you think it's justified to make it easy only for you.
But if you justify that it is useful for others as well, and nicely describe it in a request, there are already things we added betwen 3.0.0 and 3.0.4 that we got convinced are useful for many users.
But it's up to you to describe it in the way that will be convincing. All in your hands.
GitHub link: https://github.com/apache/airflow/discussions/49997#discussioncomment-14038689
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] ExternalTaskSensor (only for last dag runs) in Airflow 3 [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
GitHub user gopidesupavan added a comment to the discussion: ExternalTaskSensor (only for last dag runs) in Airflow 3
@Frost3281 you cant access db directly from workers.
GitHub link: https://github.com/apache/airflow/discussions/49997#discussioncomment-14035368
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] ExternalTaskSensor (only for last dag runs) in Airflow 3 [airflow]
Posted by "ajbosco (via GitHub)" <gi...@apache.org>.
GitHub user ajbosco added a comment to the discussion: ExternalTaskSensor (only for last dag runs) in Airflow 3
I tried `get_ti_count` takes a list of `logical_dates` but not a start/end date.
GitHub link: https://github.com/apache/airflow/discussions/49997#discussioncomment-14036653
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org
Re: [D] ExternalTaskSensor (only for last dag runs) in Airflow 3 [airflow]
Posted by "zerodarkzone (via GitHub)" <gi...@apache.org>.
GitHub user zerodarkzone added a comment to the discussion: ExternalTaskSensor (only for last dag runs) in Airflow 3
Hi, I was doing something similar in airflow 2. This is a pretty big breaking change, sesnsors like 'ExternalTaskSensor' are pretty useless without being able to get the last execution date of a dag. Is there an easy way to solve this?
GitHub link: https://github.com/apache/airflow/discussions/49997#discussioncomment-13866121
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org