ferruzzi commented on code in PR #65392:
URL: https://github.com/apache/airflow/pull/65392#discussion_r3097163587
##########
airflow-core/src/airflow/executors/base_executor.py:
##########
@@ -435,7 +438,7 @@ def fail(self, key: TaskInstanceKey, info=None) -> None:
"""
self.change_state(key, TaskInstanceState.FAILED, info)
Review Comment:
This is close, but we need to add a quick isinstance check in each of these.
As written, if called with a Callback, this will return a TaskInstanceState...
which will technically work, but only by accident. Since both
TaskInstanceState.FAILED and CallbackState.FAILED are both enums using the
string value "FAILED" they are pythonically equal. To be technically correct,
we need something like
```python
def fail(self, key: WorkloadKey, info=None) -> None:
state = TaskInstanceState.FAILED if isinstance(key, TaskInstanceKey)
else CallbackState.FAILED
self.change_state(key, state, info)
```
and adjust the type hint on `change_state`'s state parameter to
WorkloadState accordingly.
And the same for the below.
OR if you prefer (and I think maybe I do prefer....) add a
`_detect_state_enum(key)` helper (maybe think of a better name) which returns
`type[TaskInstanceState] | type[CallbackState]" using similar logic:
```python
def _detect_state_enum(key: WorkloadKey) -> type[TaskInstanceState] |
type[CallbackState]:
if isinstance(key, TaskInstanceKey):
return TaskInstanceState
return CallbackState
```
and change the call in each of these methods to`self.change_state(key,
_detect_state_enum(key).FAILED, info)`, etc. That may be cleaner and easier to
add new workload types later....
--
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]
