amoghrajesh commented on code in PR #72127:
URL: https://github.com/apache/airflow/pull/72127#discussion_r3966923696
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -751,12 +751,25 @@ def get(self, key: str, default: JsonValue = None) ->
JsonValue:
"""Return the stored value, or ``default`` if the key does not
exist."""
from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+ resp = SUPERVISOR_COMMS.send(self._build_get_message(key))
+ return self._extract_get_response(resp, key, default)
+
+ async def aget(self, key: str, default: JsonValue = None) -> JsonValue:
+ """Async version of :meth:`get` that awaits instead of blocking the
event loop."""
Review Comment:
```suggestion
"""Async version of `get` that awaits instead of blocking the event
loop."""
```
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -780,6 +793,15 @@ def set(self, key: str, value: JsonValue) -> None:
"""Write or overwrite the value for the given key. ``value`` must not
be ``None``."""
from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+ SUPERVISOR_COMMS.send(self._build_set_message(key, value))
+
+ async def aset(self, key: str, value: JsonValue) -> None:
+ """Async version of :meth:`set` that awaits instead of blocking the
event loop."""
Review Comment:
```suggestion
"""Async version of `set` that awaits instead of blocking the event
loop."""
```
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -808,39 +830,62 @@ def set(self, key: str, value: JsonValue) -> None:
msg = SetAssetStateStoreByName(name=self._name, key=key,
value=stored)
elif self._uri:
msg = SetAssetStateStoreByUri(uri=self._uri, key=key, value=stored)
- SUPERVISOR_COMMS.send(msg)
+ return msg
def delete(self, key: str) -> None:
"""Delete a single key. No-op if the key does not exist."""
- from airflow.sdk._shared.state import AssetScope
from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
- msg: ToSupervisor
- if self._name:
- msg = DeleteAssetStateStoreByName(name=self._name, key=key)
- elif self._uri:
- msg = DeleteAssetStateStoreByUri(uri=self._uri, key=key)
# DB ref first: if backend cleanup fails after this, the ref is gone
and
# deterministic keys are recoverable on next set().
- SUPERVISOR_COMMS.send(msg)
+ SUPERVISOR_COMMS.send(self._build_delete_message(key))
backend = _get_worker_state_store_backend()
if backend is not None:
backend.delete(AssetScope(name=self._name, uri=self._uri), key)
+ async def adelete(self, key: str) -> None:
+ """Async version of :meth:`delete` that awaits instead of blocking the
event loop."""
Review Comment:
```suggestion
"""Async version of `delete` that awaits instead of blocking the
event loop."""
```
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -903,18 +949,34 @@ def get(self, key: str, default: JsonValue = None) ->
JsonValue:
"""Return the stored value for the single-inlet or single-outlet task,
or ``default`` if not found."""
return self._single_accessor().get(key, default)
+ async def aget(self, key: str, default: JsonValue = None) -> JsonValue:
+ """Async version of :meth:`get` that awaits instead of blocking the
event loop."""
+ return await self._single_accessor().aget(key, default)
+
def set(self, key: str, value: JsonValue) -> None:
"""Write or overwrite the value for the single-inlet task."""
self._single_accessor().set(key, value)
+ async def aset(self, key: str, value: JsonValue) -> None:
+ """Async version of :meth:`set` that awaits instead of blocking the
event loop."""
Review Comment:
```suggestion
"""Async version of `set` that awaits instead of blocking the event
loop."""
```
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -903,18 +949,34 @@ def get(self, key: str, default: JsonValue = None) ->
JsonValue:
"""Return the stored value for the single-inlet or single-outlet task,
or ``default`` if not found."""
return self._single_accessor().get(key, default)
+ async def aget(self, key: str, default: JsonValue = None) -> JsonValue:
+ """Async version of :meth:`get` that awaits instead of blocking the
event loop."""
Review Comment:
```suggestion
"""Async version of `get` that awaits instead of blocking the event
loop."""
```
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -903,18 +949,34 @@ def get(self, key: str, default: JsonValue = None) ->
JsonValue:
"""Return the stored value for the single-inlet or single-outlet task,
or ``default`` if not found."""
return self._single_accessor().get(key, default)
+ async def aget(self, key: str, default: JsonValue = None) -> JsonValue:
+ """Async version of :meth:`get` that awaits instead of blocking the
event loop."""
+ return await self._single_accessor().aget(key, default)
+
def set(self, key: str, value: JsonValue) -> None:
"""Write or overwrite the value for the single-inlet task."""
self._single_accessor().set(key, value)
+ async def aset(self, key: str, value: JsonValue) -> None:
+ """Async version of :meth:`set` that awaits instead of blocking the
event loop."""
+ await self._single_accessor().aset(key, value)
+
def delete(self, key: str) -> None:
"""Delete a single key for the single-inlet task."""
self._single_accessor().delete(key)
+ async def adelete(self, key: str) -> None:
+ """Async version of :meth:`delete` that awaits instead of blocking the
event loop."""
Review Comment:
```suggestion
"""Async version of `delete` that awaits instead of blocking the
event loop."""
```
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -808,39 +830,62 @@ def set(self, key: str, value: JsonValue) -> None:
msg = SetAssetStateStoreByName(name=self._name, key=key,
value=stored)
elif self._uri:
msg = SetAssetStateStoreByUri(uri=self._uri, key=key, value=stored)
- SUPERVISOR_COMMS.send(msg)
+ return msg
def delete(self, key: str) -> None:
"""Delete a single key. No-op if the key does not exist."""
- from airflow.sdk._shared.state import AssetScope
from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
- msg: ToSupervisor
- if self._name:
- msg = DeleteAssetStateStoreByName(name=self._name, key=key)
- elif self._uri:
- msg = DeleteAssetStateStoreByUri(uri=self._uri, key=key)
# DB ref first: if backend cleanup fails after this, the ref is gone
and
# deterministic keys are recoverable on next set().
- SUPERVISOR_COMMS.send(msg)
+ SUPERVISOR_COMMS.send(self._build_delete_message(key))
backend = _get_worker_state_store_backend()
if backend is not None:
backend.delete(AssetScope(name=self._name, uri=self._uri), 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(self._build_delete_message(key))
+ backend = _get_worker_state_store_backend()
+ if backend is not None:
+ await backend.adelete(AssetScope(name=self._name, uri=self._uri),
key)
+
+ def _build_delete_message(self, key: str) -> ToSupervisor:
+ msg: ToSupervisor
+ if self._name:
+ msg = DeleteAssetStateStoreByName(name=self._name, key=key)
+ elif self._uri:
+ msg = DeleteAssetStateStoreByUri(uri=self._uri, key=key)
+ return msg
+
def clear(self) -> None:
"""Delete all state keys for this asset."""
- from airflow.sdk._shared.state import AssetScope
from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+ # DB ref first, same ordering rationale as delete().
+ SUPERVISOR_COMMS.send(self._build_clear_message())
+ backend = _get_worker_state_store_backend()
+ if backend is not None:
+ backend.clear(AssetScope(name=self._name, uri=self._uri))
+
+ async def aclear(self) -> None:
+ """Async version of :meth:`clear` that awaits instead of blocking the
event loop."""
Review Comment:
```suggestion
"""Async version of `clear` that awaits instead of blocking the
event loop."""
```
##########
task-sdk/tests/task_sdk/execution_time/test_context.py:
##########
@@ -1896,6 +1896,170 @@ def test_set_warns_when_value_exceeds_limit(self,
mock_supervisor_comms):
assert "max_value_storage_bytes" in
mock_log.warning.call_args[0][0]
mock_supervisor_comms.send.assert_called_once()
+ @pytest.mark.asyncio
+ async def test_aget_returns_value(self, mock_supervisor_comms):
+ """aget awaits asend and returns the stored value, without touching
sync send."""
+ mock_supervisor_comms.asend.return_value =
AssetStateStoreResult(value="2026-04-30T00:00:00Z")
+
+ result = await
AssetStateStoreAccessor(name=self.ASSET_NAME).aget("watermark")
+
+ assert result == "2026-04-30T00:00:00Z"
+ mock_supervisor_comms.asend.assert_called_once_with(
+ GetAssetStateStoreByName(name=self.ASSET_NAME, key="watermark")
+ )
+ mock_supervisor_comms.send.assert_not_called()
+
+ @pytest.mark.asyncio
+ async def test_aget_by_uri(self, mock_supervisor_comms):
+ mock_supervisor_comms.asend.return_value =
AssetStateStoreResult(value="2026-04-30T00:00:00Z")
+
+ result = await
AssetStateStoreAccessor(uri=self.ASSET_URI).aget("watermark")
+
+ assert result == "2026-04-30T00:00:00Z"
+ mock_supervisor_comms.asend.assert_called_once_with(
+ GetAssetStateStoreByUri(uri=self.ASSET_URI, key="watermark")
+ )
+
+ @pytest.mark.asyncio
+ async def test_aget_returns_default_when_key_missing(self,
mock_supervisor_comms):
+ mock_supervisor_comms.asend.return_value = ErrorResponse(
+ error=ErrorType.ASSET_STORE_NOT_FOUND, detail={"key": "watermark"}
+ )
+
+ result = await AssetStateStoreAccessor(name=self.ASSET_NAME).aget(
+ "watermark", default="2026-01-01T00:00:00+00:00"
+ )
+
+ assert result == "2026-01-01T00:00:00+00:00"
+
+ @pytest.mark.asyncio
+ async def test_aget_raises_on_error(self, mock_supervisor_comms):
+ mock_supervisor_comms.asend.return_value = ErrorResponse(
+ error=ErrorType.GENERIC_ERROR, detail={"message": "server error"}
+ )
+
+ with pytest.raises(AirflowRuntimeError):
+ await
AssetStateStoreAccessor(name=self.ASSET_NAME).aget("some_key")
+
+ @pytest.mark.asyncio
+ async def test_aget_with_custom_backend_removes_decoration_marker(self,
mock_supervisor_comms):
+ """aget unwraps the external Store marker and resolves the ref via the
backend."""
+ mock_supervisor_comms.asend.return_value = AssetStateStoreResult(
+ value=_wrap_external_ref("s3://bucket/assets/orders/watermark")
+ )
+
+ backend = MagicMock(spec=BaseStoreBackend)
+ backend.deserialize_asset_state_store_from_ref.return_value =
"2026-05-01"
+
+ with patch(
+
"airflow.sdk.execution_time.context._get_worker_state_store_backend",
return_value=backend
+ ):
+ result = await
AssetStateStoreAccessor(name=self.ASSET_NAME).aget("watermark")
+
+ assert result == "2026-05-01"
+ backend.deserialize_asset_state_store_from_ref.assert_called_once_with(
+ "s3://bucket/assets/orders/watermark"
+ )
+
+ @pytest.mark.asyncio
+ async def test_aset_operation(self, mock_supervisor_comms):
+ mock_supervisor_comms.asend.return_value = OKResponse(ok=True)
+
+ await AssetStateStoreAccessor(name=self.ASSET_NAME).aset("watermark",
"2026-04-30T00:00:00Z")
+
+ mock_supervisor_comms.asend.assert_called_once_with(
+ SetAssetStateStoreByName(name=self.ASSET_NAME, key="watermark",
value="2026-04-30T00:00:00Z")
+ )
+ mock_supervisor_comms.send.assert_not_called()
+
+ @pytest.mark.asyncio
+ async def test_aset_by_uri(self, mock_supervisor_comms):
+ mock_supervisor_comms.asend.return_value = OKResponse(ok=True)
+
+ await AssetStateStoreAccessor(uri=self.ASSET_URI).aset("watermark",
"2026-04-30T00:00:00Z")
+
+ mock_supervisor_comms.asend.assert_called_once_with(
+ SetAssetStateStoreByUri(uri=self.ASSET_URI, key="watermark",
value="2026-04-30T00:00:00Z")
+ )
+
+ @pytest.mark.asyncio
+ async def test_aset_none_raises(self, mock_supervisor_comms):
+ with pytest.raises(ValueError, match="Cannot set value as None"):
+ await
AssetStateStoreAccessor(name=self.ASSET_NAME).aset("watermark", None)
+
+ mock_supervisor_comms.asend.assert_not_called()
+
+ @pytest.mark.asyncio
+ async def test_aset_with_custom_backend_decorates_value_with_marker(self,
mock_supervisor_comms):
+ """aset wraps the custom backend ref in the external Store marker
before sending."""
+ mock_supervisor_comms.asend.return_value = OKResponse(ok=True)
+
+ backend = MagicMock(spec=BaseStoreBackend)
+ backend.serialize_asset_state_store_to_ref.return_value =
"s3://bucket/assets/orders/watermark"
+
+ with patch(
+
"airflow.sdk.execution_time.context._get_worker_state_store_backend",
return_value=backend
+ ):
+ await
AssetStateStoreAccessor(name=self.ASSET_NAME).aset("watermark", "2026-05-01")
+
+ mock_supervisor_comms.asend.assert_called_once_with(
+ SetAssetStateStoreByName(
+ name=self.ASSET_NAME,
+ key="watermark",
+
value=_wrap_external_ref("s3://bucket/assets/orders/watermark"),
+ )
+ )
+
+ @pytest.mark.asyncio
+ async def test_adelete_awaits_asend(self, mock_supervisor_comms):
+ """adelete awaits asend without touching sync send."""
+ mock_supervisor_comms.asend.return_value = OKResponse(ok=True)
+
+ await
AssetStateStoreAccessor(name=self.ASSET_NAME).adelete("watermark")
+
+ mock_supervisor_comms.asend.assert_called_once_with(
+ DeleteAssetStateStoreByName(name=self.ASSET_NAME, key="watermark")
+ )
+ mock_supervisor_comms.send.assert_not_called()
+
+ @pytest.mark.asyncio
+ async def test_aclear_awaits_asend(self, mock_supervisor_comms):
+ """aclear awaits asend without touching sync send."""
+ mock_supervisor_comms.asend.return_value = OKResponse(ok=True)
+
+ await AssetStateStoreAccessor(uri=self.ASSET_URI).aclear()
+
+
mock_supervisor_comms.asend.assert_called_once_with(ClearAssetStateStoreByUri(uri=self.ASSET_URI))
Review Comment:
```suggestion
mock_supervisor_comms.asend.assert_called_once_with(ClearAssetStateStoreByUri(uri=self.ASSET_URI))
```
##########
task-sdk/tests/task_sdk/execution_time/test_context.py:
##########
@@ -2059,6 +2223,57 @@ def test_outlet_alias_is_ignored(self,
mock_supervisor_comms):
assert accessors._total == 0
mock_supervisor_comms.send.assert_not_called()
+ @pytest.mark.asyncio
+ async def test_aget_single_inlet_simplified(self, mock_supervisor_comms):
+ asset = Asset(name=self.ASSET_NAME, uri=f"s3://{self.ASSET_NAME}")
+ mock_supervisor_comms.asend.return_value =
AssetStateStoreResult(value="v5")
+
+ result = await AssetStateStoreAccessors([asset]).aget("watermark")
+
+ assert result == "v5"
+ mock_supervisor_comms.asend.assert_called_once_with(
+ GetAssetStateStoreByName(name=self.ASSET_NAME, key="watermark")
+ )
+
+ @pytest.mark.asyncio
+ async def test_aset_single_inlet_simplified(self, mock_supervisor_comms):
+ asset = Asset(name=self.ASSET_NAME, uri=f"s3://{self.ASSET_NAME}")
+ mock_supervisor_comms.asend.return_value = OKResponse(ok=True)
+
+ await AssetStateStoreAccessors([asset]).aset("watermark", "2026-05-01")
+
+ mock_supervisor_comms.asend.assert_called_once_with(
+ SetAssetStateStoreByName(name=self.ASSET_NAME, key="watermark",
value="2026-05-01")
+ )
+
+ @pytest.mark.asyncio
+ async def test_adelete_single_inlet_simplified(self,
mock_supervisor_comms):
+ asset = Asset(name=self.ASSET_NAME, uri=f"s3://{self.ASSET_NAME}")
+ mock_supervisor_comms.asend.return_value = OKResponse(ok=True)
+
+ await AssetStateStoreAccessors([asset]).adelete("watermark")
+
+ mock_supervisor_comms.asend.assert_called_once_with(
+ DeleteAssetStateStoreByName(name=self.ASSET_NAME, key="watermark")
+ )
+
+ @pytest.mark.asyncio
+ async def test_aclear_single_inlet_simplified(self, mock_supervisor_comms):
+ asset = Asset(name=self.ASSET_NAME, uri=f"s3://{self.ASSET_NAME}")
+ mock_supervisor_comms.asend.return_value = OKResponse(ok=True)
+
+ await AssetStateStoreAccessors([asset]).aclear()
+
Review Comment:
```suggestion
```
##########
task-sdk/tests/task_sdk/execution_time/test_context.py:
##########
@@ -1896,6 +1896,170 @@ def test_set_warns_when_value_exceeds_limit(self,
mock_supervisor_comms):
assert "max_value_storage_bytes" in
mock_log.warning.call_args[0][0]
mock_supervisor_comms.send.assert_called_once()
+ @pytest.mark.asyncio
+ async def test_aget_returns_value(self, mock_supervisor_comms):
+ """aget awaits asend and returns the stored value, without touching
sync send."""
+ mock_supervisor_comms.asend.return_value =
AssetStateStoreResult(value="2026-04-30T00:00:00Z")
+
+ result = await
AssetStateStoreAccessor(name=self.ASSET_NAME).aget("watermark")
+
+ assert result == "2026-04-30T00:00:00Z"
+ mock_supervisor_comms.asend.assert_called_once_with(
+ GetAssetStateStoreByName(name=self.ASSET_NAME, key="watermark")
+ )
+ mock_supervisor_comms.send.assert_not_called()
+
+ @pytest.mark.asyncio
+ async def test_aget_by_uri(self, mock_supervisor_comms):
+ mock_supervisor_comms.asend.return_value =
AssetStateStoreResult(value="2026-04-30T00:00:00Z")
+
+ result = await
AssetStateStoreAccessor(uri=self.ASSET_URI).aget("watermark")
+
+ assert result == "2026-04-30T00:00:00Z"
+ mock_supervisor_comms.asend.assert_called_once_with(
+ GetAssetStateStoreByUri(uri=self.ASSET_URI, key="watermark")
+ )
+
+ @pytest.mark.asyncio
+ async def test_aget_returns_default_when_key_missing(self,
mock_supervisor_comms):
+ mock_supervisor_comms.asend.return_value = ErrorResponse(
+ error=ErrorType.ASSET_STORE_NOT_FOUND, detail={"key": "watermark"}
+ )
+
+ result = await AssetStateStoreAccessor(name=self.ASSET_NAME).aget(
+ "watermark", default="2026-01-01T00:00:00+00:00"
+ )
+
+ assert result == "2026-01-01T00:00:00+00:00"
+
+ @pytest.mark.asyncio
+ async def test_aget_raises_on_error(self, mock_supervisor_comms):
+ mock_supervisor_comms.asend.return_value = ErrorResponse(
+ error=ErrorType.GENERIC_ERROR, detail={"message": "server error"}
+ )
+
+ with pytest.raises(AirflowRuntimeError):
+ await
AssetStateStoreAccessor(name=self.ASSET_NAME).aget("some_key")
+
+ @pytest.mark.asyncio
+ async def test_aget_with_custom_backend_removes_decoration_marker(self,
mock_supervisor_comms):
+ """aget unwraps the external Store marker and resolves the ref via the
backend."""
+ mock_supervisor_comms.asend.return_value = AssetStateStoreResult(
+ value=_wrap_external_ref("s3://bucket/assets/orders/watermark")
+ )
+
+ backend = MagicMock(spec=BaseStoreBackend)
+ backend.deserialize_asset_state_store_from_ref.return_value =
"2026-05-01"
+
+ with patch(
+
"airflow.sdk.execution_time.context._get_worker_state_store_backend",
return_value=backend
+ ):
+ result = await
AssetStateStoreAccessor(name=self.ASSET_NAME).aget("watermark")
+
+ assert result == "2026-05-01"
+ backend.deserialize_asset_state_store_from_ref.assert_called_once_with(
+ "s3://bucket/assets/orders/watermark"
+ )
+
+ @pytest.mark.asyncio
+ async def test_aset_operation(self, mock_supervisor_comms):
+ mock_supervisor_comms.asend.return_value = OKResponse(ok=True)
+
+ await AssetStateStoreAccessor(name=self.ASSET_NAME).aset("watermark",
"2026-04-30T00:00:00Z")
+
+ mock_supervisor_comms.asend.assert_called_once_with(
+ SetAssetStateStoreByName(name=self.ASSET_NAME, key="watermark",
value="2026-04-30T00:00:00Z")
+ )
+ mock_supervisor_comms.send.assert_not_called()
+
+ @pytest.mark.asyncio
+ async def test_aset_by_uri(self, mock_supervisor_comms):
+ mock_supervisor_comms.asend.return_value = OKResponse(ok=True)
+
+ await AssetStateStoreAccessor(uri=self.ASSET_URI).aset("watermark",
"2026-04-30T00:00:00Z")
+
+ mock_supervisor_comms.asend.assert_called_once_with(
+ SetAssetStateStoreByUri(uri=self.ASSET_URI, key="watermark",
value="2026-04-30T00:00:00Z")
+ )
Review Comment:
Same comment here
##########
task-sdk/tests/task_sdk/execution_time/test_context.py:
##########
@@ -1896,6 +1896,170 @@ def test_set_warns_when_value_exceeds_limit(self,
mock_supervisor_comms):
assert "max_value_storage_bytes" in
mock_log.warning.call_args[0][0]
mock_supervisor_comms.send.assert_called_once()
+ @pytest.mark.asyncio
+ async def test_aget_returns_value(self, mock_supervisor_comms):
+ """aget awaits asend and returns the stored value, without touching
sync send."""
+ mock_supervisor_comms.asend.return_value =
AssetStateStoreResult(value="2026-04-30T00:00:00Z")
+
+ result = await
AssetStateStoreAccessor(name=self.ASSET_NAME).aget("watermark")
+
+ assert result == "2026-04-30T00:00:00Z"
+ mock_supervisor_comms.asend.assert_called_once_with(
+ GetAssetStateStoreByName(name=self.ASSET_NAME, key="watermark")
+ )
+ mock_supervisor_comms.send.assert_not_called()
+
+ @pytest.mark.asyncio
+ async def test_aget_by_uri(self, mock_supervisor_comms):
+ mock_supervisor_comms.asend.return_value =
AssetStateStoreResult(value="2026-04-30T00:00:00Z")
+
+ result = await
AssetStateStoreAccessor(uri=self.ASSET_URI).aget("watermark")
+
+ assert result == "2026-04-30T00:00:00Z"
+ mock_supervisor_comms.asend.assert_called_once_with(
+ GetAssetStateStoreByUri(uri=self.ASSET_URI, key="watermark")
+ )
Review Comment:
Should we just parameterise these tests?
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -903,18 +949,34 @@ def get(self, key: str, default: JsonValue = None) ->
JsonValue:
"""Return the stored value for the single-inlet or single-outlet task,
or ``default`` if not found."""
return self._single_accessor().get(key, default)
+ async def aget(self, key: str, default: JsonValue = None) -> JsonValue:
+ """Async version of :meth:`get` that awaits instead of blocking the
event loop."""
+ return await self._single_accessor().aget(key, default)
+
def set(self, key: str, value: JsonValue) -> None:
"""Write or overwrite the value for the single-inlet task."""
self._single_accessor().set(key, value)
+ async def aset(self, key: str, value: JsonValue) -> None:
+ """Async version of :meth:`set` that awaits instead of blocking the
event loop."""
+ await self._single_accessor().aset(key, value)
+
def delete(self, key: str) -> None:
"""Delete a single key for the single-inlet task."""
self._single_accessor().delete(key)
+ async def adelete(self, key: str) -> None:
+ """Async version of :meth:`delete` that awaits instead of blocking the
event loop."""
+ await self._single_accessor().adelete(key)
+
def clear(self) -> None:
"""Delete all state keys for the single-inlet task."""
self._single_accessor().clear()
+ async def aclear(self) -> None:
+ """Async version of :meth:`clear` that awaits instead of blocking the
event loop."""
Review Comment:
```suggestion
"""Async version of `clear` that awaits instead of blocking the
event loop."""
```
--
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]
