kaxil commented on code in PR #76009:
URL: https://github.com/apache/airflow/pull/76009#discussion_r4437453048
##########
airflow-core/docs/administration-and-deployment/task-and-asset-state-store.rst:
##########
@@ -128,6 +128,8 @@ Custom backends
A custom backend must subclass :class:`~airflow.sdk.state.BaseStoreBackend`
and implement its abstract methods: ``get``, ``set``, ``delete``, and ``clear``
for synchronous callers and the ``aget``, ``aset``, ``adelete``, and ``aclear``
async equivalents. Refer to :class:`~airflow.sdk.state.BaseStoreBackend` for
the full API.
+Values are passed to :meth:`~airflow.sdk.state.BaseStoreBackend.set` and
returned from :meth:`~airflow.sdk.state.BaseStoreBackend.get` as JSON encoded
strings, and are never the JSON ``null``. Reads decode them through
:meth:`~airflow.sdk.state.BaseStoreBackend.get_value` and
:meth:`~airflow.sdk.state.BaseStoreBackend.aget_value`; a backend that stores
values natively, rather than as JSON strings, can override those two to skip
the round trip.
Review Comment:
Nothing calls `aget_value` today (the Execution API reads are all sync and
go through `get_value`), so "Reads decode them through `get_value` and
`aget_value`" overstates it a bit. The override also only skips the decode on
reads, since `set` still receives a JSON string. Maybe: "The Execution API
reads values through `get_value` (`aget_value` is the async equivalent); a
backend that stores values natively can override them to skip decoding on
reads."
##########
shared/state/src/airflow_shared/state/__init__.py:
##########
@@ -269,6 +271,27 @@ async def aclear(
If ``None``, implementations manage their own async session internally.
"""
+ def get_value(self, scope: StoreScope, key: str, *, session: Session |
None = None) -> JsonValue:
+ """
+ Return the decoded value for the given key, or None if the key does
not exist.
+
+ The default implementation decodes what :meth:`get` returns. A stored
JSON ``null``
+ breaks the :meth:`set` contract, so it is treated as a missing key. A
backend that
+ stores values natively, rather than as JSON strings, can override this
to skip
+ the round trip.
+ """
+ if (value := self.get(scope, key, session=session)) is None:
Review Comment:
`MetastoreBackend.get` is `@provide_session`, and that wrapper only opens a
session when `session` isn't in kwargs. Since this always forwards
`session=session`, `MetastoreBackend().get_value(scope, key)` with the default
passes `None` straight through to `session.scalar(...)` and raises
`AttributeError`. The routes all pass a session so nothing hits it today, but
it's a public method that fails with its own defaults on the default backend.
Could we only forward `session` when it's set? A test calling `get_value` on
`MetastoreBackend` without a session would catch it; the stub backend in
`test_state.py` can't.
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_asset_state_store.py:
##########
@@ -86,6 +87,14 @@ def inactive_asset(session: Session) -> AssetModel:
class TestGetAssetStateByName:
+ def test_get_stored_null_returns_404(self, client: TestClient, asset:
AssetModel):
Review Comment:
`get_asset_state_store_by_uri` has its own copy of this check and
`TestGetAssetStateByUri` has no stored-null case, so reverting that handler
wouldn't fail anything. Worth a matching test there (or parametrizing this one
over `_BY_NAME_VALUE` / `_BY_URI_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]
