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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
