Lee-W commented on code in PR #68232:
URL: https://github.com/apache/airflow/pull/68232#discussion_r3529033779
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -636,9 +674,17 @@ def delete(self, key: str) -> None:
if backend is not None:
backend.delete(self._scope, key)
+ async def adelete(self, key: str) -> None:
+ """Async version of :meth:`delete` that awaits instead of blocking the
event loop."""
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ await SUPERVISOR_COMMS.asend(DeleteTaskStateStore(ti_id=self._ti_id,
key=key))
+ backend = _get_worker_state_store_backend()
Review Comment:
```suggestion
# cleanup the DB ref first, if backend cleanup fails after this, the
ref is gone and
# deterministic keys are recoverable on next set().
await SUPERVISOR_COMMS.asend(DeleteTaskStateStore(ti_id=self._ti_id,
key=key))
backend = _get_worker_state_store_backend()
```
nit: let's add a comment here.
I was about to say, let's move this check to the front and then see this
comment in `delete`
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -1285,11 +1305,6 @@ def __getitem__(self, key: Asset | AssetAlias |
AssetRef) -> Sequence[AssetEvent
@cache # Prevent multiple API access.
def get_previous_dagrun_success(ti_id: UUID) -> PrevSuccessfulDagRunResponse:
from airflow.sdk.execution_time import task_runner
- from airflow.sdk.execution_time.comms import (
Review Comment:
Thanks for removing these inline imports!
--
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]
