justinpakzad commented on code in PR #70771:
URL: https://github.com/apache/airflow/pull/70771#discussion_r3942491817
##########
airflow-ctl/src/airflowctl/api/operations.py:
##########
@@ -326,6 +329,39 @@ def delete_queued_event(self, dag_id: str, asset_id: str)
-> str | ServerRespons
self.client.delete(f"dags/{dag_id}/assets/{asset_id}/queuedEvents")
return asset_id
+ def list_state_store(self, asset_id: str) ->
AssetStateStoreCollectionResponse | ServerResponseError:
+ """List all state store entries for an asset."""
+ return super().execute_list(
+ path=f"/assets/{asset_id}/state-store",
data_model=AssetStateStoreCollectionResponse
+ )
+
+ def get_state_store(self, asset_id: str, key: str) ->
AssetStateStoreResponse | ServerResponseError:
+ """Get a single asset state store entry."""
+ self.response = self.client.get(f"assets/{asset_id}/state-store/{key}")
+ return
AssetStateStoreResponse.model_validate_json(self.response.content)
+
+ def set_state_store(self, asset_id: str, key: str, value: str) -> str |
ServerResponseError:
+ """Set an asset state store value. Creates or overwrites the key."""
+ try:
+ parsed_value = json.loads(value)
Review Comment:
Yes, it's intentional. The declared type in the Pydantic model (which comes
from the API) for the state store `value` field is `JsonValue` so it's meant
to accept any JSON type. This the same pattern used in the `XComOperations`
(add and edit commands). Your point about user expectation is still valid
though. I modified the help text so it's a bit more clear.
--
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]
