lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: validate values at schema paths

Browse Source
This commit is contained in:
lda
2026-07-22 11:34:18 +07:00 Verified
parent a22fe81425
commit 3b9b09871d
2 changed files with 229 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wf_api/schema_projection.py
+65 -9
View File
@@ -4,7 +4,7 @@ from collections.abc import Mapping, Sequence
from copy import deepcopy
from typing import Any
from jsonschema import Draft202012Validator, SchemaError
from jsonschema import Draft202012Validator, SchemaError, ValidationError
JsonObject = dict[str, Any]
@@ -21,6 +21,55 @@ def schema_path_exists(
return True
def schema_fragment_at_path(
schema: JsonObject,
parts: Sequence[str],
*,
label: str = "schema",
) -> JsonObject:
"""Return a self-contained selected schema fragment with local definitions."""
_check_schema(label, schema)
fragment = deepcopy(dict(_schema_at_path(schema, parts, label=label)))
_merge_definition_block(
fragment,
schema,
"$defs",
target_label=f"{label} fragment",
source_label=label,
)
_merge_definition_block(
fragment,
schema,
"definitions",
target_label=f"{label} fragment",
source_label=label,
)
_check_schema(f"{label} fragment", fragment)
return fragment
def validate_json_value_at_schema_path(
*,
schema: JsonObject,
parts: Sequence[str],
value: object,
label: str,
) -> None:
"""Validate one JSON-compatible literal against a selected schema path."""
fragment = schema_fragment_at_path(
schema,
parts,
label="capability input schema",
)
path = ".".join(parts) or "."
try:
Draft202012Validator(fragment).validate(value)
except ValidationError as exc:
raise ValueError(
f"{label} does not satisfy schema at {path!r}: {exc.message}"
) from exc
def project_schema_path_to_schema_path(
*,
target_schema: JsonObject,
@@ -30,16 +79,20 @@ def project_schema_path_to_schema_path(
allow_existing_equivalent: bool = False,
) -> JsonObject:
"""Copy one nested source subschema into a target object-property path."""
if not source_parts:
raise ValueError("source schema path must not be empty")
if not target_parts:
raise ValueError("target schema path must not be empty")
_check_schema("target_schema", target_schema)
_check_schema("source_schema", source_schema)
source_value = _schema_at_path(
source_schema,
source_parts,
label="source schema",
# An empty source path means the complete capability payload. This is
# distinct from an empty target path, which cannot be inserted into a parent.
source_value = (
source_schema
if not source_parts
else _schema_at_path(
source_schema,
source_parts,
label="source schema",
)
)
projected = deepcopy(target_schema)
@@ -263,15 +316,18 @@ def _merge_definition_block(
target_schema: JsonObject,
source_schema: JsonObject,
key: str,
*,
target_label: str = "state_schema",
source_label: str = "output_schema",
) -> None:
source_defs = source_schema.get(key)
if source_defs is None:
return
if not isinstance(source_defs, dict):
raise ValueError(f"output_schema.{key} must be an object")
raise ValueError(f"{source_label}.{key} must be an object")
target_defs = target_schema.setdefault(key, {})
if not isinstance(target_defs, dict):
raise ValueError(f"state_schema.{key} must be an object")
raise ValueError(f"{target_label}.{key} must be an object")
for name, definition in source_defs.items():
if name in target_defs and target_defs[name] != definition:
raise ValueError(f"conflicting {key}.{name}")
tests/wf_api/test_schema_projection.py
+164
View File
@@ -6,7 +6,9 @@ from wf_api.schema_projection import (
project_output_property_to_state_schema,
project_property_to_schema_path,
project_schema_path_to_schema_path,
schema_fragment_at_path,
schema_path_exists,
validate_json_value_at_schema_path,
)
@@ -218,6 +220,168 @@ def test_project_schema_path_copies_inline_nested_source() -> None:
}
def test_schema_fragment_selects_inline_nested_field() -> None:
fragment = schema_fragment_at_path(
{
"type": "object",
"properties": {
"request": {
"type": "object",
"properties": {"format": {"type": "string"}},
}
},
},
("request", "format"),
)
assert fragment == {"type": "string"}
def test_schema_fragment_preserves_defs_for_selected_reference() -> None:
fragment = schema_fragment_at_path(
{
"type": "object",
"properties": {"request": {"$ref": "#/$defs/Request"}},
"$defs": {
"Request": {
"type": "object",
"properties": {"format": {"type": "string"}},
}
},
},
("request",),
label="capability input schema",
)
assert fragment["$ref"] == "#/$defs/Request"
assert fragment["$defs"]["Request"]["properties"]["format"] == {"type": "string"}
def test_schema_fragment_accepts_whole_schema() -> None:
schema = {"type": "object", "properties": {"title": {"type": "string"}}}
assert schema_fragment_at_path(schema, ()) == schema
def test_schema_fragment_rejects_remote_selected_reference() -> None:
with pytest.raises(
ValueError,
match="unsupported reference 'https://example.com/request.json'",
):
schema_fragment_at_path(
{
"type": "object",
"properties": {"request": {"$ref": "https://example.com/request.json"}},
},
("request",),
)
@pytest.mark.parametrize(
("schema", "value"),
[
({"type": "string"}, "markdown"),
(
{
"type": "object",
"properties": {"title": {"type": "string"}},
"required": ["title"],
},
{"title": "Report"},
),
({"type": "array", "items": {"type": "integer"}}, [1, 2]),
({"type": ["string", "null"]}, None),
],
)
def test_validate_json_value_accepts_matching_literals(
schema: dict[str, object],
value: object,
) -> None:
validate_json_value_at_schema_path(
schema={
"type": "object",
"properties": {"value": schema},
},
parts=("value",),
value=value,
label="bindings[0].value",
)
def test_validate_json_value_at_nested_schema_path() -> None:
schema = {
"type": "object",
"properties": {"request": {"$ref": "#/$defs/Request"}},
"$defs": {
"Request": {
"type": "object",
"properties": {"format": {"enum": ["markdown", "json"]}},
}
},
}
validate_json_value_at_schema_path(
schema=schema,
parts=("request", "format"),
value="markdown",
label="bindings[0].value",
)
with pytest.raises(
ValueError,
match=r"bindings\[0\]\.value does not satisfy schema at 'request.format'",
):
validate_json_value_at_schema_path(
schema=schema,
parts=("request", "format"),
value="html",
label="bindings[0].value",
)
@pytest.mark.parametrize(
("schema", "value"),
[
({"type": "string"}, 7),
({"type": "object"}, []),
({"type": "array"}, {}),
({"type": "null"}, "not-null"),
],
)
def test_validate_json_value_rejects_non_matching_literals(
schema: dict[str, object],
value: object,
) -> None:
with pytest.raises(
ValueError,
match=r"bindings\[0\]\.value does not satisfy schema at 'value'",
):
validate_json_value_at_schema_path(
schema={
"type": "object",
"properties": {"value": schema},
},
parts=("value",),
value=value,
label="bindings[0].value",
)
def test_project_schema_path_accepts_whole_source_schema() -> None:
projected = project_schema_path_to_schema_path(
target_schema={"type": "object", "properties": {}},
source_schema={
"type": "object",
"properties": {"title": {"type": "string"}},
"required": ["title"],
},
source_parts=(),
target_parts=("payload",),
)
assert projected["properties"]["payload"]["required"] == ["title"]
def test_project_schema_path_traverses_pydantic_defs_reference() -> None:
projected = project_schema_path_to_schema_path(
target_schema={"type": "object", "properties": {}},