Lee-W commented on code in PR #66160:
URL: https://github.com/apache/airflow/pull/66160#discussion_r3180930467
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -406,6 +404,172 @@ def get(self, key, default: Any = NOTSET) -> Any:
raise
+class TaskStateAccessor:
+ """Accessor for task state scoped to the current task instance. Available
as ``context['task_state']`` at task execution time."""
+
+ def __init__(self, ti_id: UUID) -> None:
+ self._ti_id = ti_id
+
+ def __eq__(self, other: object) -> bool:
+ if not isinstance(other, TaskStateAccessor):
+ return False
+ return self._ti_id == other._ti_id
+
+ def __hash__(self) -> int:
+ return hash(self._ti_id)
+
+ def __repr__(self) -> str:
+ return f"<TaskStateAccessor ti_id={self._ti_id}>"
+
+ # TODO: ``__getattr__`` for jinja template access like ``{{
task_state.job_id }}``
+ # is not implemented yet cos it's unclear whether task state values will be
+ # used in templates.
+
+ def get(self, key: str) -> str | None:
+ """Return the stored value, or ``None`` if the key does not exist."""
+ from airflow.sdk.execution_time.comms import ErrorResponse,
GetTaskState, TaskStateResult
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ resp = SUPERVISOR_COMMS.send(GetTaskState(ti_id=self._ti_id, key=key))
+ if isinstance(resp, ErrorResponse):
+ if resp.error == ErrorType.TASK_STATE_NOT_FOUND:
+ return None
+ raise AirflowRuntimeError(resp)
+ if isinstance(resp, TaskStateResult):
+ return resp.value
+ return None
Review Comment:
```suggestion
if isinstance(resp, ErrorResponse) and resp.error !=
ErrorType.TASK_STATE_NOT_FOUND:
raise AirflowRuntimeError(resp)
if isinstance(resp, TaskStateResult):
return resp.value
return None
```
##########
task-sdk/src/airflow/sdk/api/client.py:
##########
@@ -639,6 +643,100 @@ def get_sequence_slice(
return XComSequenceSliceResponse.model_validate_json(resp.read())
+class TaskStateOperations:
+ __slots__ = ("client",)
+
+ def __init__(self, client: Client):
+ self.client = client
+
+ def get(self, ti_id: uuid.UUID, key: str) -> TaskStateResponse |
ErrorResponse:
+ """Get a task state value from the API server."""
+ try:
+ resp = self.client.get(f"state/ti/{ti_id}/{key}")
+ except ServerResponseError as e:
+ if e.response.status_code == HTTPStatus.NOT_FOUND:
+ log.debug("Task state key not found", ti_id=ti_id, key=key)
+ return ErrorResponse(error=ErrorType.TASK_STATE_NOT_FOUND,
detail={"key": key})
+ raise
+ return TaskStateResponse.model_validate_json(resp.read())
+
+ def set(self, ti_id: uuid.UUID, key: str, value: str) -> OKResponse:
+ """Set a task state value via the API server."""
+ body = TaskStatePutBody(value=value)
+ self.client.put(f"state/ti/{ti_id}/{key}",
content=body.model_dump_json())
+ return OKResponse(ok=True)
+
+ def delete(self, ti_id: uuid.UUID, key: str) -> OKResponse:
+ """Delete a single task state key via the API server."""
+ self.client.delete(f"state/ti/{ti_id}/{key}")
+ return OKResponse(ok=True)
+
+ def clear(self, ti_id: uuid.UUID, all_map_indices: bool = False) ->
OKResponse:
+ """Clear all task state keys for a task instance via the API server."""
+ params = {"all_map_indices": "true"} if all_map_indices else {}
+ self.client.delete(f"state/ti/{ti_id}", params=params)
+ return OKResponse(ok=True)
+
+
+class AssetStateOperations:
+ __slots__ = ("client",)
+
+ def __init__(self, client: Client):
+ self.client = client
+
+ def get(
+ self, key: str, *, name: str | None = None, uri: str | None = None
+ ) -> AssetStateResponse | ErrorResponse:
+ """Get an asset state value from the API server."""
+ if name:
+ endpoint, params = "state/asset/by-name/value", {"name": name,
"key": key}
+ elif uri:
+ endpoint, params = "state/asset/by-uri/value", {"uri": uri, "key":
key}
+ else:
+ raise ValueError("Either `name` or `uri` must be provided")
Review Comment:
This duplicate part can be extracted as a function
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -406,6 +404,172 @@ def get(self, key, default: Any = NOTSET) -> Any:
raise
+class TaskStateAccessor:
+ """Accessor for task state scoped to the current task instance. Available
as ``context['task_state']`` at task execution time."""
+
+ def __init__(self, ti_id: UUID) -> None:
+ self._ti_id = ti_id
+
+ def __eq__(self, other: object) -> bool:
+ if not isinstance(other, TaskStateAccessor):
+ return False
+ return self._ti_id == other._ti_id
+
+ def __hash__(self) -> int:
+ return hash(self._ti_id)
+
+ def __repr__(self) -> str:
+ return f"<TaskStateAccessor ti_id={self._ti_id}>"
+
+ # TODO: ``__getattr__`` for jinja template access like ``{{
task_state.job_id }}``
+ # is not implemented yet cos it's unclear whether task state values will be
+ # used in templates.
+
+ def get(self, key: str) -> str | None:
+ """Return the stored value, or ``None`` if the key does not exist."""
+ from airflow.sdk.execution_time.comms import ErrorResponse,
GetTaskState, TaskStateResult
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ resp = SUPERVISOR_COMMS.send(GetTaskState(ti_id=self._ti_id, key=key))
+ if isinstance(resp, ErrorResponse):
+ if resp.error == ErrorType.TASK_STATE_NOT_FOUND:
+ return None
+ raise AirflowRuntimeError(resp)
+ if isinstance(resp, TaskStateResult):
+ return resp.value
+ return None
+
+ def set(self, key: str, value: str) -> None:
+ """Write or overwrite the value for the given key."""
+ from airflow.sdk.execution_time.comms import SetTaskState
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ SUPERVISOR_COMMS.send(SetTaskState(ti_id=self._ti_id, key=key,
value=value))
+
+ def delete(self, key: str) -> None:
+ """Delete a single key. No-op if the key does not exist."""
+ from airflow.sdk.execution_time.comms import DeleteTaskState
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ SUPERVISOR_COMMS.send(DeleteTaskState(ti_id=self._ti_id, key=key))
+
+ def clear(self, all_map_indices: bool = False) -> None:
+ """
+ Delete all keys for this task instance.
+
+ Pass ``all_map_indices=True`` to wipe state across every mapped
+ instance of the task (fleet-wide reset). Defaults to clearing only
+ this task instance's own state.
+ """
+ from airflow.sdk.execution_time.comms import ClearTaskState
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ SUPERVISOR_COMMS.send(ClearTaskState(ti_id=self._ti_id,
all_map_indices=all_map_indices))
+
+
+class AssetStateAccessor:
+ """
+ Accessor for asset state scoped to an asset.
+
+ Available as ``context['asset_state']`` when the task has exactly one inlet
+ asset - asset watcher pattern. The accessor is bound to that Asset at
context construction time.
+ """
+
+ # TODO: When the task has zero or more than one inlet — the key
+ # will not be present in ``context`` in those cases. Multi-inlet support
+ # ie: ``context['asset_state'][MY_ASSET].get(...)`` can be developed as
needed.
+
+ def __init__(self, *, name: str | None = None, uri: str | None = None) ->
None:
+ self._name = name
+ self._uri = uri
+
+ def __eq__(self, other: object) -> bool:
+ if not isinstance(other, AssetStateAccessor):
+ return False
+ return self._name == other._name and self._uri == other._uri
+
+ def __hash__(self) -> int:
+ return hash((self._name, self._uri))
+
+ def __repr__(self) -> str:
+ if self._name is not None:
+ return f"<AssetStateAccessor name={self._name!r}>"
+ return f"<AssetStateAccessor uri={self._uri!r}>"
+
+ def get(self, key: str) -> str | None:
+ """Return the stored value, or ``None`` if the key does not exist."""
+ from airflow.sdk.execution_time.comms import (
+ AssetStateResult,
+ ErrorResponse,
+ GetAssetStateByName,
+ GetAssetStateByUri,
+ ToSupervisor,
+ )
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ msg: ToSupervisor
+ if self._name:
+ msg = GetAssetStateByName(name=self._name, key=key)
+ elif self._uri:
+ msg = GetAssetStateByUri(uri=self._uri, key=key)
+ else:
+ raise ValueError("Either `name` or `uri` must be provided")
+ resp = SUPERVISOR_COMMS.send(msg)
+ if isinstance(resp, ErrorResponse):
+ if resp.error == ErrorType.ASSET_STATE_NOT_FOUND:
+ return None
+ raise AirflowRuntimeError(resp)
+ if isinstance(resp, AssetStateResult):
+ return resp.value
+ return None
+
+ def set(self, key: str, value: str) -> None:
+ """Write or overwrite the value for the given key."""
+ from airflow.sdk.execution_time.comms import SetAssetStateByName,
SetAssetStateByUri, ToSupervisor
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ msg: ToSupervisor
+ if self._name:
+ msg = SetAssetStateByName(name=self._name, key=key, value=value)
+ elif self._uri:
+ msg = SetAssetStateByUri(uri=self._uri, key=key, value=value)
+ else:
+ raise ValueError("Either `name` or `uri` must be provided")
Review Comment:
```suggestion
msg: ToSupervisor
if self._name:
msg = OpCls(name=self._name, key=key, value=value)
elif self._uri:
msg = OpCls(uri=self._uri, key=key, value=value)
else:
raise ValueError("Either `name` or `uri` must be provided")
```
Not sure whether it's over-engineering. But this can duplicate some code.
ignore it if it looks worse
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -406,6 +404,172 @@ def get(self, key, default: Any = NOTSET) -> Any:
raise
+class TaskStateAccessor:
+ """Accessor for task state scoped to the current task instance. Available
as ``context['task_state']`` at task execution time."""
+
+ def __init__(self, ti_id: UUID) -> None:
+ self._ti_id = ti_id
+
+ def __eq__(self, other: object) -> bool:
+ if not isinstance(other, TaskStateAccessor):
+ return False
+ return self._ti_id == other._ti_id
+
+ def __hash__(self) -> int:
+ return hash(self._ti_id)
+
+ def __repr__(self) -> str:
+ return f"<TaskStateAccessor ti_id={self._ti_id}>"
+
+ # TODO: ``__getattr__`` for jinja template access like ``{{
task_state.job_id }}``
+ # is not implemented yet cos it's unclear whether task state values will be
+ # used in templates.
+
+ def get(self, key: str) -> str | None:
+ """Return the stored value, or ``None`` if the key does not exist."""
+ from airflow.sdk.execution_time.comms import ErrorResponse,
GetTaskState, TaskStateResult
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ resp = SUPERVISOR_COMMS.send(GetTaskState(ti_id=self._ti_id, key=key))
+ if isinstance(resp, ErrorResponse):
+ if resp.error == ErrorType.TASK_STATE_NOT_FOUND:
+ return None
+ raise AirflowRuntimeError(resp)
+ if isinstance(resp, TaskStateResult):
+ return resp.value
+ return None
+
+ def set(self, key: str, value: str) -> None:
+ """Write or overwrite the value for the given key."""
+ from airflow.sdk.execution_time.comms import SetTaskState
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ SUPERVISOR_COMMS.send(SetTaskState(ti_id=self._ti_id, key=key,
value=value))
+
+ def delete(self, key: str) -> None:
+ """Delete a single key. No-op if the key does not exist."""
+ from airflow.sdk.execution_time.comms import DeleteTaskState
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ SUPERVISOR_COMMS.send(DeleteTaskState(ti_id=self._ti_id, key=key))
+
+ def clear(self, all_map_indices: bool = False) -> None:
+ """
+ Delete all keys for this task instance.
+
+ Pass ``all_map_indices=True`` to wipe state across every mapped
+ instance of the task (fleet-wide reset). Defaults to clearing only
+ this task instance's own state.
+ """
+ from airflow.sdk.execution_time.comms import ClearTaskState
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ SUPERVISOR_COMMS.send(ClearTaskState(ti_id=self._ti_id,
all_map_indices=all_map_indices))
+
+
+class AssetStateAccessor:
+ """
+ Accessor for asset state scoped to an asset.
+
+ Available as ``context['asset_state']`` when the task has exactly one inlet
+ asset - asset watcher pattern. The accessor is bound to that Asset at
context construction time.
+ """
+
+ # TODO: When the task has zero or more than one inlet — the key
+ # will not be present in ``context`` in those cases. Multi-inlet support
+ # ie: ``context['asset_state'][MY_ASSET].get(...)`` can be developed as
needed.
+
+ def __init__(self, *, name: str | None = None, uri: str | None = None) ->
None:
Review Comment:
Is `None` here used to distinguish `Asset`, `Asset.ref(uri=...)` and
`Asset.ref(name=...)`? if not, we should set the value if missing.
e.g.,
`Asset(uri="uri")` is basically `Asset(uri="uri", name="uri")` in the db
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -406,6 +404,172 @@ def get(self, key, default: Any = NOTSET) -> Any:
raise
+class TaskStateAccessor:
+ """Accessor for task state scoped to the current task instance. Available
as ``context['task_state']`` at task execution time."""
+
+ def __init__(self, ti_id: UUID) -> None:
+ self._ti_id = ti_id
+
+ def __eq__(self, other: object) -> bool:
+ if not isinstance(other, TaskStateAccessor):
+ return False
+ return self._ti_id == other._ti_id
+
+ def __hash__(self) -> int:
+ return hash(self._ti_id)
+
+ def __repr__(self) -> str:
+ return f"<TaskStateAccessor ti_id={self._ti_id}>"
+
+ # TODO: ``__getattr__`` for jinja template access like ``{{
task_state.job_id }}``
+ # is not implemented yet cos it's unclear whether task state values will be
+ # used in templates.
+
+ def get(self, key: str) -> str | None:
+ """Return the stored value, or ``None`` if the key does not exist."""
+ from airflow.sdk.execution_time.comms import ErrorResponse,
GetTaskState, TaskStateResult
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ resp = SUPERVISOR_COMMS.send(GetTaskState(ti_id=self._ti_id, key=key))
+ if isinstance(resp, ErrorResponse):
+ if resp.error == ErrorType.TASK_STATE_NOT_FOUND:
+ return None
+ raise AirflowRuntimeError(resp)
+ if isinstance(resp, TaskStateResult):
+ return resp.value
+ return None
+
+ def set(self, key: str, value: str) -> None:
+ """Write or overwrite the value for the given key."""
+ from airflow.sdk.execution_time.comms import SetTaskState
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
Review Comment:
I guess moving this import to the top would cause issues?
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -406,6 +404,172 @@ def get(self, key, default: Any = NOTSET) -> Any:
raise
+class TaskStateAccessor:
+ """Accessor for task state scoped to the current task instance. Available
as ``context['task_state']`` at task execution time."""
+
+ def __init__(self, ti_id: UUID) -> None:
+ self._ti_id = ti_id
+
+ def __eq__(self, other: object) -> bool:
+ if not isinstance(other, TaskStateAccessor):
+ return False
+ return self._ti_id == other._ti_id
+
+ def __hash__(self) -> int:
+ return hash(self._ti_id)
+
+ def __repr__(self) -> str:
+ return f"<TaskStateAccessor ti_id={self._ti_id}>"
+
+ # TODO: ``__getattr__`` for jinja template access like ``{{
task_state.job_id }}``
+ # is not implemented yet cos it's unclear whether task state values will be
+ # used in templates.
+
+ def get(self, key: str) -> str | None:
+ """Return the stored value, or ``None`` if the key does not exist."""
+ from airflow.sdk.execution_time.comms import ErrorResponse,
GetTaskState, TaskStateResult
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ resp = SUPERVISOR_COMMS.send(GetTaskState(ti_id=self._ti_id, key=key))
+ if isinstance(resp, ErrorResponse):
+ if resp.error == ErrorType.TASK_STATE_NOT_FOUND:
+ return None
+ raise AirflowRuntimeError(resp)
+ if isinstance(resp, TaskStateResult):
+ return resp.value
+ return None
+
+ def set(self, key: str, value: str) -> None:
+ """Write or overwrite the value for the given key."""
+ from airflow.sdk.execution_time.comms import SetTaskState
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ SUPERVISOR_COMMS.send(SetTaskState(ti_id=self._ti_id, key=key,
value=value))
+
+ def delete(self, key: str) -> None:
+ """Delete a single key. No-op if the key does not exist."""
+ from airflow.sdk.execution_time.comms import DeleteTaskState
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ SUPERVISOR_COMMS.send(DeleteTaskState(ti_id=self._ti_id, key=key))
+
+ def clear(self, all_map_indices: bool = False) -> None:
+ """
+ Delete all keys for this task instance.
+
+ Pass ``all_map_indices=True`` to wipe state across every mapped
+ instance of the task (fleet-wide reset). Defaults to clearing only
+ this task instance's own state.
+ """
+ from airflow.sdk.execution_time.comms import ClearTaskState
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ SUPERVISOR_COMMS.send(ClearTaskState(ti_id=self._ti_id,
all_map_indices=all_map_indices))
+
+
+class AssetStateAccessor:
+ """
+ Accessor for asset state scoped to an asset.
+
+ Available as ``context['asset_state']`` when the task has exactly one inlet
+ asset - asset watcher pattern. The accessor is bound to that Asset at
context construction time.
+ """
+
+ # TODO: When the task has zero or more than one inlet — the key
+ # will not be present in ``context`` in those cases. Multi-inlet support
+ # ie: ``context['asset_state'][MY_ASSET].get(...)`` can be developed as
needed.
+
+ def __init__(self, *, name: str | None = None, uri: str | None = None) ->
None:
Review Comment:
this kinda affect how `__repr__` works
##########
task-sdk/src/airflow/sdk/execution_time/context.py:
##########
@@ -406,6 +404,172 @@ def get(self, key, default: Any = NOTSET) -> Any:
raise
+class TaskStateAccessor:
+ """Accessor for task state scoped to the current task instance. Available
as ``context['task_state']`` at task execution time."""
+
+ def __init__(self, ti_id: UUID) -> None:
+ self._ti_id = ti_id
+
+ def __eq__(self, other: object) -> bool:
+ if not isinstance(other, TaskStateAccessor):
+ return False
+ return self._ti_id == other._ti_id
+
+ def __hash__(self) -> int:
+ return hash(self._ti_id)
+
+ def __repr__(self) -> str:
+ return f"<TaskStateAccessor ti_id={self._ti_id}>"
+
+ # TODO: ``__getattr__`` for jinja template access like ``{{
task_state.job_id }}``
+ # is not implemented yet cos it's unclear whether task state values will be
+ # used in templates.
+
+ def get(self, key: str) -> str | None:
+ """Return the stored value, or ``None`` if the key does not exist."""
+ from airflow.sdk.execution_time.comms import ErrorResponse,
GetTaskState, TaskStateResult
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ resp = SUPERVISOR_COMMS.send(GetTaskState(ti_id=self._ti_id, key=key))
+ if isinstance(resp, ErrorResponse):
+ if resp.error == ErrorType.TASK_STATE_NOT_FOUND:
+ return None
+ raise AirflowRuntimeError(resp)
+ if isinstance(resp, TaskStateResult):
+ return resp.value
+ return None
+
+ def set(self, key: str, value: str) -> None:
+ """Write or overwrite the value for the given key."""
+ from airflow.sdk.execution_time.comms import SetTaskState
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ SUPERVISOR_COMMS.send(SetTaskState(ti_id=self._ti_id, key=key,
value=value))
+
+ def delete(self, key: str) -> None:
+ """Delete a single key. No-op if the key does not exist."""
+ from airflow.sdk.execution_time.comms import DeleteTaskState
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ SUPERVISOR_COMMS.send(DeleteTaskState(ti_id=self._ti_id, key=key))
+
+ def clear(self, all_map_indices: bool = False) -> None:
+ """
+ Delete all keys for this task instance.
+
+ Pass ``all_map_indices=True`` to wipe state across every mapped
+ instance of the task (fleet-wide reset). Defaults to clearing only
+ this task instance's own state.
+ """
+ from airflow.sdk.execution_time.comms import ClearTaskState
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ SUPERVISOR_COMMS.send(ClearTaskState(ti_id=self._ti_id,
all_map_indices=all_map_indices))
+
+
+class AssetStateAccessor:
+ """
+ Accessor for asset state scoped to an asset.
+
+ Available as ``context['asset_state']`` when the task has exactly one inlet
+ asset - asset watcher pattern. The accessor is bound to that Asset at
context construction time.
+ """
+
+ # TODO: When the task has zero or more than one inlet — the key
+ # will not be present in ``context`` in those cases. Multi-inlet support
+ # ie: ``context['asset_state'][MY_ASSET].get(...)`` can be developed as
needed.
+
+ def __init__(self, *, name: str | None = None, uri: str | None = None) ->
None:
+ self._name = name
+ self._uri = uri
+
+ def __eq__(self, other: object) -> bool:
+ if not isinstance(other, AssetStateAccessor):
+ return False
+ return self._name == other._name and self._uri == other._uri
+
+ def __hash__(self) -> int:
+ return hash((self._name, self._uri))
+
+ def __repr__(self) -> str:
+ if self._name is not None:
+ return f"<AssetStateAccessor name={self._name!r}>"
+ return f"<AssetStateAccessor uri={self._uri!r}>"
+
+ def get(self, key: str) -> str | None:
+ """Return the stored value, or ``None`` if the key does not exist."""
+ from airflow.sdk.execution_time.comms import (
+ AssetStateResult,
+ ErrorResponse,
+ GetAssetStateByName,
+ GetAssetStateByUri,
+ ToSupervisor,
+ )
+ from airflow.sdk.execution_time.task_runner import SUPERVISOR_COMMS
+
+ msg: ToSupervisor
+ if self._name:
+ msg = GetAssetStateByName(name=self._name, key=key)
+ elif self._uri:
+ msg = GetAssetStateByUri(uri=self._uri, key=key)
+ else:
+ raise ValueError("Either `name` or `uri` must be provided")
+ resp = SUPERVISOR_COMMS.send(msg)
+ if isinstance(resp, ErrorResponse):
+ if resp.error == ErrorType.ASSET_STATE_NOT_FOUND:
+ return None
+ raise AirflowRuntimeError(resp)
+ if isinstance(resp, AssetStateResult):
+ return resp.value
Review Comment:
```suggestion
if isinstance(resp, ErrorResponse) and resp.error !=
ErrorType.ASSET_STATE_NOT_FOUND:
raise AirflowRuntimeError(resp)
if isinstance(resp, AssetStateResult):
return resp.value
```
--
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]
