mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-22 21:44:18 +02:00
Handle allOf and oneOf in json_schema_to_type
Previously these compositions silently resolved to Any, disabling
all validation for schemas that use them. allOf is now merged into
a single object schema; oneOf creates a Union type.
🤖 Generated with Claude Code
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
William Easton
strawgate
parent
ee48a0fd6e
commit
e4d9c24922
2 changed files with 280 additions and 1 deletions
|
|
@ -453,7 +453,7 @@ def _schema_to_type(
|
|||
if not schema:
|
||||
return object
|
||||
|
||||
if "type" not in schema and "properties" in schema:
|
||||
if "type" not in schema and "properties" in schema and "allOf" not in schema:
|
||||
return _create_dataclass(schema, schema.get("title", "<unknown>"), schemas)
|
||||
|
||||
# Handle references first
|
||||
|
|
@ -493,6 +493,79 @@ def _schema_to_type(
|
|||
else:
|
||||
return Union[tuple(types)] # type: ignore # noqa: UP007
|
||||
|
||||
# Handle allOf (schema intersection / composition).
|
||||
# Flatten all sub-schemas into a single merged object schema.
|
||||
if "allOf" in schema:
|
||||
merged: dict[str, Any] = {}
|
||||
merged_properties: dict[str, Any] = {}
|
||||
merged_required: list[str] = []
|
||||
|
||||
def _collect_allof(sub: Any) -> None:
|
||||
"""Recursively collect properties from a sub-schema."""
|
||||
if isinstance(sub, bool):
|
||||
return
|
||||
if "$ref" in sub:
|
||||
sub = dict(_resolve_ref(sub["$ref"], schemas))
|
||||
# Recurse into nested allOf
|
||||
if "allOf" in sub:
|
||||
for nested in sub["allOf"]:
|
||||
_collect_allof(nested)
|
||||
merged_properties.update(sub.get("properties", {}))
|
||||
merged_required.extend(sub.get("required", []))
|
||||
for key in ("title", "description", "additionalProperties"):
|
||||
if key in sub and key not in merged:
|
||||
merged[key] = sub[key]
|
||||
|
||||
# Include sibling properties/required from the schema itself,
|
||||
# not just from allOf children. This handles schemas like
|
||||
# {"properties": {"local": ...}, "allOf": [{"properties": {"inherited": ...}}]}
|
||||
merged_properties.update(schema.get("properties", {}))
|
||||
merged_required.extend(schema.get("required", []))
|
||||
|
||||
for sub in schema["allOf"]:
|
||||
_collect_allof(sub)
|
||||
if merged_properties:
|
||||
merged["type"] = "object"
|
||||
merged["properties"] = merged_properties
|
||||
if merged_required:
|
||||
merged["required"] = list(dict.fromkeys(merged_required))
|
||||
return _schema_to_type(merged, schemas)
|
||||
# allOf with no mergeable properties — fall through to Any
|
||||
|
||||
# Handle oneOf (exactly-one-of union).
|
||||
# Treat like anyOf for type construction — Pydantic's Union
|
||||
# does "first match" which is a reasonable approximation.
|
||||
if "oneOf" in schema:
|
||||
types: list[type | Any] = []
|
||||
for subschema in schema["oneOf"]:
|
||||
# Same dict special-case as the anyOf handler: detect
|
||||
# map-like objects so they become dict[str, X] instead
|
||||
# of empty dataclasses that discard key/value data.
|
||||
if (
|
||||
isinstance(subschema, dict)
|
||||
and subschema.get("type") == "object"
|
||||
and not subschema.get("properties")
|
||||
and subschema.get("additionalProperties")
|
||||
):
|
||||
additional_props = subschema["additionalProperties"]
|
||||
if additional_props is True:
|
||||
types.append(dict[str, Any])
|
||||
else:
|
||||
value_type = _schema_to_type(additional_props, schemas)
|
||||
types.append(dict[str, value_type]) # type: ignore
|
||||
else:
|
||||
types.append(_schema_to_type(subschema, schemas))
|
||||
has_null = type(None) in types
|
||||
types = [t for t in types if t is not type(None)]
|
||||
if len(types) == 0:
|
||||
return type(None)
|
||||
elif len(types) == 1:
|
||||
return types[0] | None if has_null else types[0] # type: ignore
|
||||
else:
|
||||
if has_null:
|
||||
return Union[(*types, type(None))] # type: ignore
|
||||
return Union[tuple(types)] # type: ignore # noqa: UP007
|
||||
|
||||
schema_type = schema.get("type")
|
||||
if not schema_type:
|
||||
return Any
|
||||
|
|
|
|||
|
|
@ -404,3 +404,209 @@ class TestUnsupportedPatternFallback:
|
|||
warnings.simplefilter("error", UserWarning)
|
||||
T = json_schema_to_type(schema) # must not warn
|
||||
TypeAdapter(T).validate_python("hello")
|
||||
|
||||
class TestAllOfOneOf:
|
||||
"""allOf and oneOf composition, previously returning Any."""
|
||||
|
||||
def test_allof_merges_properties(self):
|
||||
"""allOf sub-schemas should be merged into a single object type."""
|
||||
schema = {
|
||||
"allOf": [
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {"name": {"type": "string"}},
|
||||
"required": ["name"],
|
||||
},
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {"age": {"type": "integer"}},
|
||||
},
|
||||
]
|
||||
}
|
||||
T = json_schema_to_type(schema)
|
||||
ta = TypeAdapter(T)
|
||||
|
||||
result = ta.validate_python({"name": "Alice", "age": 30})
|
||||
assert result.name == "Alice" # ty:ignore[unresolved-attribute]
|
||||
assert result.age == 30 # ty:ignore[unresolved-attribute]
|
||||
|
||||
def test_allof_preserves_required(self):
|
||||
"""Required fields from all allOf sub-schemas should be enforced."""
|
||||
schema = {
|
||||
"allOf": [
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {"name": {"type": "string"}},
|
||||
"required": ["name"],
|
||||
},
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {"age": {"type": "integer"}},
|
||||
},
|
||||
]
|
||||
}
|
||||
T = json_schema_to_type(schema)
|
||||
ta = TypeAdapter(T)
|
||||
|
||||
with pytest.raises(ValidationError):
|
||||
ta.validate_python({"age": 30}) # missing required 'name'
|
||||
|
||||
def test_allof_with_ref(self):
|
||||
"""allOf with $ref should resolve and merge the referenced schema."""
|
||||
schema = {
|
||||
"allOf": [
|
||||
{"$ref": "#/$defs/Base"},
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {"extra": {"type": "string"}},
|
||||
},
|
||||
],
|
||||
"$defs": {
|
||||
"Base": {
|
||||
"type": "object",
|
||||
"properties": {"id": {"type": "integer"}},
|
||||
"required": ["id"],
|
||||
}
|
||||
},
|
||||
}
|
||||
T = json_schema_to_type(schema)
|
||||
ta = TypeAdapter(T)
|
||||
|
||||
result = ta.validate_python({"id": 1, "extra": "hello"})
|
||||
assert result.id == 1 # ty:ignore[unresolved-attribute]
|
||||
assert result.extra == "hello" # ty:ignore[unresolved-attribute]
|
||||
|
||||
def test_oneof_creates_union(self):
|
||||
"""oneOf should create a Union type that accepts any sub-schema."""
|
||||
schema = {
|
||||
"oneOf": [
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"kind": {"const": "dog"},
|
||||
"breed": {"type": "string"},
|
||||
},
|
||||
"required": ["kind", "breed"],
|
||||
},
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"kind": {"const": "cat"},
|
||||
"indoor": {"type": "boolean"},
|
||||
},
|
||||
"required": ["kind", "indoor"],
|
||||
},
|
||||
]
|
||||
}
|
||||
T = json_schema_to_type(schema)
|
||||
ta = TypeAdapter(T)
|
||||
|
||||
dog = ta.validate_python({"kind": "dog", "breed": "lab"})
|
||||
assert dog.kind == "dog" # ty:ignore[unresolved-attribute]
|
||||
|
||||
cat = ta.validate_python({"kind": "cat", "indoor": True})
|
||||
assert cat.indoor is True # ty:ignore[unresolved-attribute]
|
||||
|
||||
def test_oneof_with_scalars(self):
|
||||
"""oneOf with scalar types should create a Union."""
|
||||
schema = {
|
||||
"oneOf": [
|
||||
{"type": "string"},
|
||||
{"type": "integer"},
|
||||
]
|
||||
}
|
||||
T = json_schema_to_type(schema)
|
||||
ta = TypeAdapter(T)
|
||||
|
||||
assert ta.validate_python("hello") == "hello"
|
||||
assert ta.validate_python(42) == 42
|
||||
|
||||
def test_nested_allof(self):
|
||||
"""allOf inside allOf should be flattened recursively."""
|
||||
schema = {
|
||||
"allOf": [
|
||||
{
|
||||
"allOf": [
|
||||
{"type": "object", "properties": {"a": {"type": "string"}}},
|
||||
{"type": "object", "properties": {"b": {"type": "integer"}}},
|
||||
]
|
||||
},
|
||||
{"type": "object", "properties": {"c": {"type": "boolean"}}},
|
||||
]
|
||||
}
|
||||
T = json_schema_to_type(schema)
|
||||
ta = TypeAdapter(T)
|
||||
result = ta.validate_python({"a": "x", "b": 1, "c": True})
|
||||
assert result.a == "x" # ty:ignore[unresolved-attribute]
|
||||
assert result.b == 1 # ty:ignore[unresolved-attribute]
|
||||
assert result.c is True # ty:ignore[unresolved-attribute]
|
||||
|
||||
def test_allof_ref_to_allof(self):
|
||||
"""allOf with $ref pointing to another allOf should resolve fully."""
|
||||
schema = {
|
||||
"allOf": [
|
||||
{"$ref": "#/$defs/Combined"},
|
||||
{"type": "object", "properties": {"extra": {"type": "string"}}},
|
||||
],
|
||||
"$defs": {
|
||||
"Combined": {
|
||||
"allOf": [
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {"x": {"type": "integer"}},
|
||||
"required": ["x"],
|
||||
},
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {"y": {"type": "integer"}},
|
||||
"required": ["y"],
|
||||
},
|
||||
]
|
||||
}
|
||||
},
|
||||
}
|
||||
T = json_schema_to_type(schema)
|
||||
ta = TypeAdapter(T)
|
||||
result = ta.validate_python({"x": 1, "y": 2, "extra": "hi"})
|
||||
assert result.x == 1 # ty:ignore[unresolved-attribute]
|
||||
assert result.y == 2 # ty:ignore[unresolved-attribute]
|
||||
assert result.extra == "hi" # ty:ignore[unresolved-attribute]
|
||||
|
||||
def test_allof_with_sibling_properties(self):
|
||||
"""Sibling properties on the same schema as allOf should be included."""
|
||||
schema = {
|
||||
"properties": {"local": {"type": "string"}},
|
||||
"required": ["local"],
|
||||
"allOf": [
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {"inherited": {"type": "integer"}},
|
||||
}
|
||||
],
|
||||
}
|
||||
T = json_schema_to_type(schema)
|
||||
ta = TypeAdapter(T)
|
||||
|
||||
result = ta.validate_python({"local": "hi", "inherited": 42})
|
||||
assert result.local == "hi" # ty:ignore[unresolved-attribute]
|
||||
assert result.inherited == 42 # ty:ignore[unresolved-attribute]
|
||||
|
||||
with pytest.raises(ValidationError):
|
||||
ta.validate_python({"inherited": 42}) # missing required 'local'
|
||||
|
||||
def test_oneof_with_dict_branch(self):
|
||||
"""oneOf with a map-like object branch should produce dict[str, X]."""
|
||||
schema = {
|
||||
"oneOf": [
|
||||
{"type": "string"},
|
||||
{
|
||||
"type": "object",
|
||||
"additionalProperties": {"type": "integer"},
|
||||
},
|
||||
]
|
||||
}
|
||||
T = json_schema_to_type(schema)
|
||||
ta = TypeAdapter(T)
|
||||
|
||||
assert ta.validate_python("hello") == "hello"
|
||||
assert ta.validate_python({"x": 1, "y": 2}) == {"x": 1, "y": 2}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue