jason810496 commented on code in PR #71054:
URL: https://github.com/apache/airflow/pull/71054#discussion_r3974863182
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/common.py:
##########
@@ -32,6 +34,56 @@
log = logging.getLogger(__name__)
+
+def find_reserved_keys(
+ value: Any,
+ reserved_keys: Collection[str],
Review Comment:
Let's use `FORBIDDEN_XCOM_KEYS` for this function and no need to introduce
the `SERDE_RESERVED_DICT_KEYS`. Since `FORBIDDEN_XCOM_KEYS` is wider than the
`SERDE_RESERVED_DICT_KEYS`.
```suggestion
```
##########
airflow-core/docs/tutorial/hitl.rst:
##########
@@ -221,6 +221,12 @@ calls involved (``~`` works as a wildcard for ``dag_id``
and ``dag_run_id``):
PATCH
/api/v2/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}/{map_index}/hitlDetails
{"chosen_options": ["Approve"], "params_input": {}}
+.. note::
+
+ Keys in ``params_input`` may not be Airflow's reserved serialization keys
(``__classname__`` or
Review Comment:
```suggestion
Keys in ``params_input`` must not be Airflow's reserved serialization
keys (``__classname__`` or
```
##########
shared/serialization/src/airflow_shared/serialization/__init__.py:
##########
@@ -42,6 +42,11 @@
}
)
+# serde.serialize refuses to serialize a dict that contains either of these
keys. This is narrower
+# than FORBIDDEN_XCOM_KEYS above, which also blocks keys that only matter when
XCom reads a value
+# back and decodes it into an object.
+SERDE_RESERVED_DICT_KEYS = frozenset({CLASSNAME, SCHEMA_ID})
+
Review Comment:
```suggestion
```
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/hitl.py:
##########
@@ -33,6 +35,21 @@ class UpdateHITLDetailPayload(BaseModel):
chosen_options: list[str] = Field(min_length=1)
params_input: Mapping = Field(default_factory=dict)
+ @field_validator("params_input")
+ @classmethod
+ def _check_serde_reserved_keys(cls, params_input: Mapping) -> Mapping:
+ # serde.serialize refuses these keys, and it only runs once the task
resumes, long after the
+ # response was stored and the request returned. Reject it here
instead, while the user can
+ # still correct the input and resubmit.
+ found = find_reserved_keys(params_input, SERDE_RESERVED_DICT_KEYS,
root="params_input")
Review Comment:
Then please remember to update this downstream function.
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/common.py:
##########
@@ -32,6 +34,56 @@
log = logging.getLogger(__name__)
+
+def find_reserved_keys(
+ value: Any,
+ reserved_keys: Collection[str],
+ *,
+ root: str = "value",
+ decode_json_strings: bool = False,
+) -> tuple[str, list[str]] | None:
+ """
+ Find the first mapping under ``value`` that holds one of ``reserved_keys``.
+
+ Returns the dotted path to that mapping and the reserved keys it holds, or
``None`` if there
+ are none.
+
+ Write paths that take arbitrary user data use this to turn away bad input
while there is still
+ a request to fail. Storing it instead pushes the failure into whatever
reads the value back,
+ where the error surfaces far from the person who submitted it.
+
+ Pass ``decode_json_strings`` when the read path parses stored strings back
into containers, the
+ way XCom's does. Without it, a payload sent as ``json.dumps({...})`` slips
through and comes
+ back out as a dict holding the reserved key. Leave it off where a string
stays a string, as
+ with ``serde.serialize``.
+ """
+ reserved = frozenset(reserved_keys)
+
+ def walk(obj: Any, path: str) -> tuple[str, list[str]] | None:
+ if isinstance(obj, str):
+ if not decode_json_strings:
+ return None
+ try:
+ decoded = json.loads(obj)
+ except (ValueError, TypeError):
+ return None
+ return walk(decoded, path) if isinstance(decoded, (dict, list))
else None
+ if isinstance(obj, Mapping):
+ found = reserved & obj.keys()
Review Comment:
```suggestion
def walk(obj: Any, path: str) -> tuple[str, list[str]] | None:
if isinstance(obj, str):
if not decode_json_strings:
return None
try:
decoded = json.loads(obj)
except (ValueError, TypeError):
return None
return walk(decoded, path) if isinstance(decoded, (dict, list))
else None
if isinstance(obj, Mapping):
found = FORBIDDEN_XCOM_KEYS & obj.keys()
```
--
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]
