You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2021/07/15 12:11:06 UTC
[GitHub] [airflow] tsabsch opened a new issue #17022: TaskInstance.previous_execution_date_success fails with: 'TaskInstance' object has no attribute 'task'
tsabsch opened a new issue #17022:
URL: https://github.com/apache/airflow/issues/17022
**Apache Airflow version**: 1.10.15
**Kubernetes version**: 1.18.17-gke.1901
**Environment**:
- **Cloud provider or hardware configuration**: GCP Cloud Composer (composer-1.16.6-airflow-1.10.15)
- **OS** (e.g. from /etc/os-release):
- **Kernel** (e.g. `uname -a`):
- **Install tools**: Terraform google provider
- **Others**:
**What happened**:
I tried to get the last successful execution date for a given TaskInstance object which I retrieved via `dag_run.get_task_instances()`, but got the following error:
```
Traceback (most recent call last)
File "/usr/local/lib/airflow/airflow/models/taskinstance.py", line 985, in _run_raw_task
result = task_copy.execute(context=context)
File "/usr/local/lib/airflow/airflow/operators/python_operator.py", line 113, in execute
return_value = self.execute_callable()
File "/usr/local/lib/airflow/airflow/operators/python_operator.py", line 118, in execute_callable
return self.python_callable(*self.op_args, **self.op_kwargs)
File "/home/airflow/gcs/dags/tsabsch_understand_retries.py", line 8, in some_function
dates = [ti.previous_execution_date_success for ti in tis]
File "/home/airflow/gcs/dags/tsabsch_understand_retries.py", line 8, in <listcomp>
dates = [ti.previous_execution_date_success for ti in tis]
File "/usr/local/lib/airflow/airflow/models/taskinstance.py", line 625, in previous_execution_date_success
prev_ti = self._get_previous_ti(state=State.SUCCESS)
File "/usr/local/lib/airflow/airflow/utils/db.py", line 74, in wrapper
return func(*args, **kwargs)
File "/usr/local/lib/airflow/airflow/models/taskinstance.py", line 582, in _get_previous_ti
dag = self.task.dag
AttributeError: 'TaskInstance' object has no attribute 'task'
```
**What you expected to happen**:
To get the last successful execution date for a given TaskInstance object.
**How to reproduce it**:
Originally I want to use this as part of a custom `on_failure_notification`, but for the sake of MWE I created a normal PythonOperator with the same behaviour:
```python
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.utils.dates import days_ago
def some_function(**kwargs):
tis = kwargs["dag_run"].get_task_instances()
dates = [ti.previous_execution_date_success for ti in tis]
print(dates)
dag = DAG("mwe", start_date=days_ago(1))
op = PythonOperator(
task_id='mwe_task',
python_callable=some_function,
provide_context=True,
dag=dag,
)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal closed issue #17022: TaskInstance.previous_execution_date_success fails with: 'TaskInstance' object has no attribute 'task'
Posted by GitBox <gi...@apache.org>.
eladkal closed issue #17022:
URL: https://github.com/apache/airflow/issues/17022
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] eladkal commented on issue #17022: TaskInstance.previous_execution_date_success fails with: 'TaskInstance' object has no attribute 'task'
Posted by GitBox <gi...@apache.org>.
eladkal commented on issue #17022:
URL: https://github.com/apache/airflow/issues/17022#issuecomment-906452189
As stated by others above the functionality still exist:
Instead of using `previous_execution_date_success`
you can do `get_previous_execution_date(state=State.SUCCESS)`
https://github.com/apache/airflow/blob/bfdda083d684b279b6b07e79080373ae4d36939a/airflow/models/taskinstance.py#L945-L949
It was changed in https://github.com/apache/airflow/pull/7900
I'm closing as there is no action item on this issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] josh-fell commented on issue #17022: TaskInstance.previous_execution_date_success fails with: 'TaskInstance' object has no attribute 'task'
Posted by GitBox <gi...@apache.org>.
josh-fell commented on issue #17022:
URL: https://github.com/apache/airflow/issues/17022#issuecomment-887959994
FWIW this is also reproducible in Airflow 2.
**Airflow version**
2.1.0
Using the DAG below, the `get_previous_execution_date()` call from `TaskInstance` object directly succeeds while the same method call accessing the `TaskInstance` from the `DagRun` object fails for the same exception noted originally.
```python
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.utils.dates import days_ago
dag = DAG("mwe", start_date=days_ago(14))
def some_function(dag_run, ti):
from airflow.utils.state import State
dag_run_tis = dag_run.get_task_instances()
print(ti.get_previous_execution_date(state=State.SUCCESS))
print(dag_run_tis[0].get_previous_execution_date(state=State.SUCCESS))
with dag:
op = PythonOperator(task_id="op_some_function", python_callable=some_function)
```
It appears as though the the `TaskInstance` objects return different sets of attributes when accessing them directly via the "ti" context or the "dag_run" context and indeed, there is no `task` attribute in the `TaskInstance` returned via `DagRun`.
Get Previous Execution Date from TaskInstance | Get Previous Execution Date via DagRun |
----------------------------------------------- | ----------------------------------------- |
`ti.get_previous_execution_date()` | `dag_run.get_task_instances()[0].get_previous_execution_date()` |
_log | _sa_instance_state |
_sa_instance_state | _try_number |
_try_number | dag_id |
dag_id | duration |
duration | end_date |
end_date | execution_date |
execution_date | executor_config |
executor_config | external_executor_id |
hostname | hostname |
job_id | job_id |
max_tries | max_tries |
operator | operator |
pid | pid |
pool | pool |
pool_slots | pool_slots |
priority_weight | priority_weight |
queue | queue |
queued_dttm | queued_by_job_id |
raw | queued_dttm |
run_as_user | start_date |
start_date | state |
state | task_id |
task | test_mode |
task_id | unixname |
test_mode |
unixname |
@tsabsch Have you tried directly accessing the `TaskInstance` rather than going through `DagRun` as a workaround?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] cansjt edited a comment on issue #17022: TaskInstance.previous_execution_date_success fails with: 'TaskInstance' object has no attribute 'task'
Posted by GitBox <gi...@apache.org>.
cansjt edited a comment on issue #17022:
URL: https://github.com/apache/airflow/issues/17022#issuecomment-891787393
With Airflow 2.1.2 we are something similar but not identical:
```
Task failed with exception", "exc_info": "Traceback (most recent call last):
File \"/usr/local/lib/python3.8/site-packages/airflow/models/taskinstance.py\", line 1157, in _run_raw_task
self._prepare_and_execute_task_with_callbacks(context, task)
File \"/usr/local/lib/python3.8/site-packages/airflow/models/taskinstance.py\", line 1331, in _prepare_and_execute_task_with_callbacks
result = self._execute_task(context, task_copy)
File \"/usr/local/lib/python3.8/site-packages/airflow/models/taskinstance.py\", line 1361, in _execute_task
result = task_copy.execute(context=context)
File \"/usr/local/lib/python3.8/site-packages/airflow/operators/python.py\", line 150, in execute
return_value = self.execute_callable()
File \"/usr/local/lib/python3.8/site-packages/airflow/operators/python.py\", line 161, in execute_callable
return self.python_callable(*self.op_args, **self.op_kwargs)
File \"/usr/local/lib/python3.8/site-packages/aggregator/tasks/airflow.py\", line 58, in __call__
super().__call__(execution_date + timedelta(days=1), **context)
File \"/usr/local/lib/python3.8/site-packages/aggregator/tasks/__init__.py\", line 103, in __call__
start_date, end_date = self._period(execution_date, **context)
File \"/usr/local/lib/python3.8/site-packages/aggregator/retention.py\", line 189, in __call__
lower_bounds, upper_bounds = zip(*[w(execution_date, **context) for w in self._windows])
File \"/usr/local/lib/python3.8/site-packages/aggregator/retention.py\", line 189, in <listcomp>
lower_bounds, upper_bounds = zip(*[w(execution_date, **context) for w in self._windows])
File \"/usr/local/lib/python3.8/site-packages/aggregator/tasks/airflow.py\", line 77, in __call__
if task_instance is None or task_instance.previous_execution_date_success is None:
AttributeError: 'TaskInstance' object has no attribute 'previous_execution_date_success'
```
The code triggering the execution is this
```python
class FromLastRunWindow(TimeWindow):
def __init__(self, *, clamp: bool = True, **kwargs):
super().__init__(**kwargs)
self._clamp = clamp
def __call__(self, execution_date: datetime, **context) -> Tuple[datetime, datetime]:
upper = datetime.max
if self._clamp:
upper = self._fix_date(execution_date)
task_instance = context.get('task_instance')
if task_instance is None or task_instance.previous_execution_date_success is None: # Raises here
return datetime.min, upper
return datetime.combine(task_instance.previous_execution_date_success.date(),
time(0, 0, 0, 0)), upper
```
This is an auxiliary function we use in various tasks, as follows:
```python
def python_operator_callable(execution_date, **context):
start, end = FromLastRunWindow()(execution_date, **context)
# Proceed with the tasks operations ...
```
Works fine with 1.10.15. Looking at the documentation (versions 1.10.15, 2.0.0 & 2.12) it seems the `TaskInstance.previous_execution_date_success` has simply been removed, without any deprecation notice. And the proper way to retrieve the same info is the new API @josh-fell used in is example (the `TaskInstance.get_previous_execution_date()` method)
So the question is more will this be a *wontfix*, forcing anyone depending on this to update their code or is it worth a PR ? Something more or less like:
```py
class TaskInstance(...):
@property
def previous_execution_date_success(self):
"""The execution_date of the instance of this of this task from latest succesful dag run.
.. deprecated:: 2.1.3
This property will be removed in Airflow 2.y.0
"""
warnings.warn('The `TasksInstance.previous_execution_date_success` has been deprecated '
'and will be removed in Airflow 2.y.0. Use '
'TaskInstance.get_previous_execution_date(state=State.SUCCESS)` instead.',
DeprecationWarning)
return self.get_previous_execution_date(state=State.SUCCESS)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] cansjt commented on issue #17022: TaskInstance.previous_execution_date_success fails with: 'TaskInstance' object has no attribute 'task'
Posted by GitBox <gi...@apache.org>.
cansjt commented on issue #17022:
URL: https://github.com/apache/airflow/issues/17022#issuecomment-891787393
With Airflow 2.1.2 we are something similar but not identical:
```
Task failed with exception", "exc_info": "Traceback (most recent call last):
File \"/usr/local/lib/python3.8/site-packages/airflow/models/taskinstance.py\", line 1157, in _run_raw_task
self._prepare_and_execute_task_with_callbacks(context, task)
File \"/usr/local/lib/python3.8/site-packages/airflow/models/taskinstance.py\", line 1331, in _prepare_and_execute_task_with_callbacks
result = self._execute_task(context, task_copy)
File \"/usr/local/lib/python3.8/site-packages/airflow/models/taskinstance.py\", line 1361, in _execute_task
result = task_copy.execute(context=context)
File \"/usr/local/lib/python3.8/site-packages/airflow/operators/python.py\", line 150, in execute
return_value = self.execute_callable()
File \"/usr/local/lib/python3.8/site-packages/airflow/operators/python.py\", line 161, in execute_callable
return self.python_callable(*self.op_args, **self.op_kwargs)
File \"/usr/local/lib/python3.8/site-packages/aggregator/tasks/airflow.py\", line 58, in __call__
super().__call__(execution_date + timedelta(days=1), **context)
File \"/usr/local/lib/python3.8/site-packages/aggregator/tasks/__init__.py\", line 103, in __call__
start_date, end_date = self._period(execution_date, **context)
File \"/usr/local/lib/python3.8/site-packages/aggregator/retention.py\", line 189, in __call__
lower_bounds, upper_bounds = zip(*[w(execution_date, **context) for w in self._windows])
File \"/usr/local/lib/python3.8/site-packages/aggregator/retention.py\", line 189, in <listcomp>
lower_bounds, upper_bounds = zip(*[w(execution_date, **context) for w in self._windows])
File \"/usr/local/lib/python3.8/site-packages/aggregator/tasks/airflow.py\", line 77, in __call__
if task_instance is None or task_instance.previous_execution_date_success is None:
AttributeError: 'TaskInstance' object has no attribute 'previous_execution_date_success'
```
The code triggering the execution is this
```python
class FromLastRunWindow(TimeWindow):
def __init__(self, *, clamp: bool = True, **kwargs):
super().__init__(**kwargs)
self._clamp = clamp
def __call__(self, execution_date: datetime, **context) -> Tuple[datetime, datetime]:
upper = datetime.max
if self._clamp:
upper = self._fix_date(execution_date)
task_instance = context.get('task_instance')
if task_instance is None or task_instance.previous_execution_date_success is None: # Raises here
return datetime.min, upper
return datetime.combine(task_instance.previous_execution_date_success.date(),
time(0, 0, 0, 0)), upper
```
This is an auxiliary function we use in various tasks, as follows:
```python
def python_operator_callable(execution_date, **context):
start, end = FromLastRunWindow()(execution_date, **context)
# Proceed with the tasks operations ...
```
Works fine with 1.10.15. Looking at the documentation (versions 1.10.15, 2.0.0 & 2.12) it seems the `TaskInstance.previous_execution_date_success` as simply been removed, without any deprecation notice. And the proper way to retrieve the same info is the new API @josh-fell used in is example (the `TaskInstance.get_previous_execution_date()` method)
So the question is more will this be a *wontfix*, forcing anyone depending on this to update their code or is it worth a PR ? Something more or less like:
```py
class TaskInstance(...):
@property
def previous_execution_date_success(self):
"""The execution_date of the instance of this of this task from latest succesful dag run.
.. deprecated:: 2.1.3
This property will be removed in Airflow 2.y.0
"""
warnings.warn('The `TasksInstance.previous_execution_date_success` has been deprecated '
'and will be removed in Airflow 2.y.0. Use '
'TaskInstance.get_previous_execution_date(state=State.SUCCESS)` instead.',
DeprecationWarning)
return self.get_previous_execution_date(state=State.SUCCESS)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] tsabsch commented on issue #17022: TaskInstance.previous_execution_date_success fails with: 'TaskInstance' object has no attribute 'task'
Posted by GitBox <gi...@apache.org>.
tsabsch commented on issue #17022:
URL: https://github.com/apache/airflow/issues/17022#issuecomment-889967622
Thanks for the reply @josh-fell. Yes, I can confirm that using the context's `ti` directly works. That covers however only a subset of scenarios, because it only works for the `TaskInstance` directly associated with the operator, not the other `TaskInstance`s in the `DagRun`. In my actual use case, I want to use the code in the `on_failure_callback` to gather statistics about the `Dag`when a run fails. There, the `context["ti"]` always refers to the first `TaskInstance` in the `DagRun`.
Digging a bit into the code, the difference appears to be that `dag_run.get_task_instances()` retrieves the `TaskInstance`s by a lookup from the SQL database [1], which returns only the fields defined in the SQLAlchemy model as opposed to a `TaskInstance` that was instantiated through its `__init__` [3]. I found in the class docstring, that this is a conscious design decision. In this case, I think the documentation could need some more information about the limitations of `dag_run.get_task_instances()`.
[1] https://github.com/apache/airflow/blob/v1-10-stable/airflow/models/dagrun.py#L171
[2] https://github.com/apache/airflow/blob/v1-10-stable/airflow/models/taskinstance.py#L182
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org