This is an automated email from the ASF dual-hosted git repository.
wenjin272 pushed a commit to branch release-0.3
in repository https://gitbox.apache.org/repos/asf/flink-agents.git
The following commit(s) were added to refs/heads/release-0.3 by this push:
new 58fd13d1 [bug] [python] Propagate config model reconstruction errors
in Action deserialize (#921)
58fd13d1 is described below
commit 58fd13d1ccc94a69db1cb07e362f5f4733f474a9
Author: Alan Z. <[email protected]>
AuthorDate: Sat Jul 25 04:29:58 2026 -0700
[bug] [python] Propagate config model reconstruction errors in Action
deserialize (#921)
(cherry picked from commit 4d995672b1cbde606d2799130e7bf27e650f8169)
---
python/flink_agents/plan/actions/action.py | 28 +++++++-----
.../flink_agents/plan/tests/resources/action.json | 11 ++---
python/flink_agents/plan/tests/test_action.py | 51 ++++++++++++++++++++++
3 files changed, 75 insertions(+), 15 deletions(-)
diff --git a/python/flink_agents/plan/actions/action.py
b/python/flink_agents/plan/actions/action.py
index 96f2f5f0..6376433a 100644
--- a/python/flink_agents/plan/actions/action.py
+++ b/python/flink_agents/plan/actions/action.py
@@ -26,6 +26,9 @@ from flink_agents.api.runner_context import RunnerContext
from flink_agents.plan.function import Function, JavaFunction, PythonFunction
_CONFIG_TYPE = "__config_type__"
+# Tags a config entry that we serialized from a pydantic model, so on the way
+# back we only rebuild the ones we tagged and leave plain values alone.
+_PYDANTIC_MODEL_MARKER = "__pydantic_model__"
class Action(BaseModel):
@@ -60,11 +63,12 @@ class Action(BaseModel):
data[_CONFIG_TYPE] = "python"
for name, value in config.items():
if isinstance(value, BaseModel):
- data[name] = (
- inspect.getmodule(value).__name__,
- value.__class__.__name__,
- value,
- )
+ data[name] = {
+ _PYDANTIC_MODEL_MARKER: True,
+ "module": inspect.getmodule(value).__name__,
+ "class": value.__class__.__name__,
+ "value": value,
+ }
else:
data[name] = value
return data
@@ -85,11 +89,15 @@ class Action(BaseModel):
self["config"][name] = value["value"]
return self
for name, value in config.items():
- try:
- module = importlib.import_module(value[0])
- clazz = getattr(module, value[1])
- self["config"][name] = clazz.model_validate(value[2])
- except Exception: # noqa : PERF203
+ # Rebuild only entries with our marker, so a plain list or dict
+ # value is not treated as a model by mistake. Do not catch errors:
+ # if the class cannot be imported (e.g. missing on the worker),
+ # fail here with the real cause instead of a confusing error later.
+ if isinstance(value, dict) and value.get(_PYDANTIC_MODEL_MARKER):
+ module = importlib.import_module(value["module"])
+ clazz = getattr(module, value["class"])
+ self["config"][name] = clazz.model_validate(value["value"])
+ else:
self["config"][name] = value
return self
diff --git a/python/flink_agents/plan/tests/resources/action.json
b/python/flink_agents/plan/tests/resources/action.json
index 7ec09d6c..4db36767 100644
--- a/python/flink_agents/plan/tests/resources/action.json
+++ b/python/flink_agents/plan/tests/resources/action.json
@@ -10,10 +10,11 @@
],
"config": {
"__config_type__": "python",
- "output_schema": [
- "flink_agents.api.agents.types",
- "OutputSchema",
- {
+ "output_schema": {
+ "__pydantic_model__": true,
+ "module": "flink_agents.api.agents.types",
+ "class": "OutputSchema",
+ "value": {
"output_schema": {
"names": [
"result"
@@ -23,6 +24,6 @@
]
}
}
- ]
+ }
}
}
\ No newline at end of file
diff --git a/python/flink_agents/plan/tests/test_action.py
b/python/flink_agents/plan/tests/test_action.py
index 5a75f718..9d9045c8 100644
--- a/python/flink_agents/plan/tests/test_action.py
+++ b/python/flink_agents/plan/tests/test_action.py
@@ -120,3 +120,54 @@ def
test_action_deserialize_java_shape_config_unwraps_primitives() -> None:
"rate": 1.5,
"label": "fast",
}
+
+
+def test_action_deserialize_python_config_propagates_reconstruction_error() ->
None:
+ """A tagged model whose module is missing must fail loudly.
+
+ The entry is marked as a serialized model, so importing a non-existent
+ module raises instead of being silently swallowed.
+ """
+ json_str = json.dumps(
+ {
+ "name": "legal",
+ "exec": {
+ "func_type": "PythonFunction",
+ "module": "flink_agents.plan.tests.test_action",
+ "qualname": "legal_signature",
+ },
+ "trigger_conditions": ["_input_event"],
+ "config": {
+ "__config_type__": "python",
+ "broken": {
+ "__pydantic_model__": True,
+ "module": "nonexistent_module_xyz",
+ "class": "SomeClass",
+ "value": {},
+ },
+ },
+ }
+ )
+ with pytest.raises(ModuleNotFoundError):
+ Action.model_validate_json(json_str)
+
+
+def test_action_deserialize_python_config_preserves_plain_list() -> None:
+ """A plain user list must survive untouched, not be mistaken for a
model."""
+ json_str = json.dumps(
+ {
+ "name": "legal",
+ "exec": {
+ "func_type": "PythonFunction",
+ "module": "flink_agents.plan.tests.test_action",
+ "qualname": "legal_signature",
+ },
+ "trigger_conditions": ["_input_event"],
+ "config": {
+ "__config_type__": "python",
+ "hosts": ["host-a", "host-b", "host-c"],
+ },
+ }
+ )
+ action = Action.model_validate_json(json_str)
+ assert action.config == {"hosts": ["host-a", "host-b", "host-c"]}
