Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 31 additions & 1 deletion python/composio/utils/shared.py
Original file line number Diff line number Diff line change
Expand Up @@ -587,13 +587,22 @@ def demo_function(
# that are missing a "type" key or use an unrecognized type, then build
# a Union for any count of members (no 1/2/3-member cap).
mapped_types: t.List[t.Any] = [
PYDANTIC_TYPE_TO_PYTHON_TYPE.get(ptype, t.Any) for ptype in param_types
_annotation_from_json_schema_type(ptype) for ptype in param_types
]
if len(mapped_types) == 1:
annotation = mapped_types[0]
else:
annotation = reduce(lambda a, b: t.Union[a, b], mapped_types)
param_default = param_schema.get("default", "")
elif isinstance(param_type, list):
annotation = _annotation_from_json_schema_type(param_type)
scalar_type = param_type[0] if len(param_type) == 1 else None
if isinstance(scalar_type, str) and scalar_type in FALLBACK_VALUES:
param_default = param_schema.get(
"default", FALLBACK_VALUES[scalar_type]
)
else:
param_default = param_schema.get("default", "")
elif param_type in PYDANTIC_TYPE_TO_PYTHON_TYPE:
annotation = PYDANTIC_TYPE_TO_PYTHON_TYPE[param_type]
param_default = param_schema.get("default", FALLBACK_VALUES[param_type])
Expand Down Expand Up @@ -625,6 +634,27 @@ def demo_function(
return default_parameters + none_default_parameters


def _annotation_from_json_schema_type(schema_type: t.Any) -> t.Any:
"""Convert a JSON Schema ``type`` value to a Python annotation.

JSON Schema Draft 2020-12 and OpenAPI 3.1 allow ``type`` to be an array,
including inside ``anyOf`` and ``oneOf`` branches. Keep that form from
reaching the scalar dictionary lookup, where a list would be unhashable.
"""
if isinstance(schema_type, list):
mapped_types = [_annotation_from_json_schema_type(item) for item in schema_type]
if not mapped_types:
return t.Any
if len(mapped_types) == 1:
return mapped_types[0]
return reduce(lambda left, right: t.Union[left, right], mapped_types)

if isinstance(schema_type, str):
return PYDANTIC_TYPE_TO_PYTHON_TYPE.get(schema_type, t.Any)

return t.Any


def get_pydantic_signature_format_from_schema_params(
schema_params: t.Dict,
skip_default: bool = False,
Expand Down
83 changes: 81 additions & 2 deletions python/tests/test_schema_parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -1496,10 +1496,14 @@ class TestGetSignatureFormatFromSchemaParams:
"""Test cases for get_signature_format_from_schema_params union handling."""

@staticmethod
def _annotation(schema):
def _parameter(schema):
params = get_signature_format_from_schema_params(schema)
assert len(params) == 1
return params[0].annotation
return params[0]

@classmethod
def _annotation(cls, schema):
return cls._parameter(schema).annotation

@pytest.mark.unit
@pytest.mark.schema
Expand Down Expand Up @@ -1631,6 +1635,81 @@ def test_anyof_with_null_member_still_resolves(self):
assert str in args
assert type(None) in args

@pytest.mark.unit
@pytest.mark.schema
@pytest.mark.parametrize(
("schema_type", "expected_members"),
[
(["string", "null"], {str, type(None)}),
(["integer"], {int}),
(["string", "integer"], {str, int}),
(["file", "null"], {t.Any, type(None)}),
],
)
def test_type_list_resolves(self, schema_type, expected_members):
"""List-valued JSON Schema types produce an annotation without raising."""
annotation = self._annotation({"properties": {"value": {"type": schema_type}}})
actual = set(t.get_args(annotation)) if t.get_args(annotation) else {annotation}
assert expected_members <= actual

@pytest.mark.unit
@pytest.mark.schema
@pytest.mark.parametrize(
("schema_type", "expected_annotation", "expected_default"),
[
(["integer"], int, 0),
(["boolean"], bool, False),
(["array"], t.List, []),
],
)
def test_single_type_list_uses_scalar_fallback(
self, schema_type, expected_annotation, expected_default
):
"""A one-item type list keeps its scalar type's implicit default."""
parameter = self._parameter({"properties": {"value": {"type": schema_type}}})

assert parameter.annotation is expected_annotation
assert parameter.default == expected_default

@pytest.mark.unit
@pytest.mark.schema
def test_single_type_list_preserves_explicit_default(self):
"""An explicit default overrides the fallback for a one-item type list."""
parameter = self._parameter(
{"properties": {"value": {"type": ["integer"], "default": 42}}}
)

assert parameter.default == 42

@pytest.mark.unit
@pytest.mark.schema
def test_multi_type_list_keeps_union_fallback(self):
"""A multi-type list remains a union with the existing empty-string fallback."""
parameter = self._parameter(
{"properties": {"value": {"type": ["integer", "null"]}}}
)

assert {int, type(None)} <= set(t.get_args(parameter.annotation))
assert parameter.default == ""

@pytest.mark.unit
@pytest.mark.schema
def test_combiner_option_with_type_list_resolves(self):
"""List-valued types inside combiners do not reach a scalar dict lookup."""
schema = {
"properties": {
"value": {
"anyOf": [
{"type": ["string", "null"]},
{"type": "integer"},
]
}
}
}

annotation = self._annotation(schema)
assert {str, int, type(None)} <= set(t.get_args(annotation))


if __name__ == "__main__":
pytest.main([__file__, "-v"])
Loading