""" JSON Schema to Pydantic type conversion using json-schema-to-pydantic library. This module provides a wrapper around json-schema-to-pydantic that maintains backwards compatibility with the existing json_schema_to_pydantic_type() API. The library handles most JSON Schema features correctly, but doesn't support boolean schemas (true/false values that are valid in JSON Schema draft-06+). We handle those by pre-filtering them before passing to the library. """ import typing as t from functools import reduce from json_schema_to_pydantic import ( CombinerError, SchemaError, create_model as create_model_from_schema, ) from composio.utils.logging import get as get_logger logger = get_logger(__name__) # Type mapping for simple cases where we don't need full model creation PYDANTIC_TYPE_TO_PYTHON_TYPE = { "string": str, "integer": int, "number": float, "boolean": bool, "array": t.List, "object": t.Dict, "null": t.Optional[t.Any], } CONTAINER_TYPE = ("array", "object") # Should be deprecated, # required values will always be provided by users # Non-required values are nullable(None) if default value not provided. FALLBACK_VALUES = { "string": "", "number": 0.0, "integer": 0, "boolean": False, "object": {}, "array": [], "null": None, } def _filter_boolean_schemas( schema: t.Union[t.Dict[str, t.Any], bool, t.List], ) -> t.Union[t.Dict[str, t.Any], t.Any, None]: """ Pre-filter boolean schemas from anyOf/allOf/oneOf arrays. JSON Schema draft-06+ allows `true` and `false` as valid schemas: - `true` means "accept any value" (equivalent to {}) - `false` means "reject all values" (equivalent to {"not": {}}) The json-schema-to-pydantic library doesn't handle these, so we: - Replace `true` with {} (empty schema, accepts anything) - Filter out `false` (rejects everything, so no point including it) Returns None if the schema is a standalone `false` boolean. """ if isinstance(schema, bool): if schema: # true -> empty schema (accepts any value) return {} else: # false -> reject all, return None to filter out return None if isinstance(schema, list): # Filter list items (e.g., for anyOf arrays) filtered = [] for item in schema: result = _filter_boolean_schemas(item) if result is not None: filtered.append(result) return filtered if filtered else None if not isinstance(schema, dict): return schema # Make a copy to avoid mutating the original result = {} for key, value in schema.items(): if key in ("anyOf", "allOf", "oneOf"): # Filter boolean schemas from combiner arrays filtered_value = _filter_boolean_schemas(value) if filtered_value is None or ( isinstance(filtered_value, list) and len(filtered_value) == 0 ): # All schemas were false, skip this combiner continue result[key] = filtered_value elif key == "items" and isinstance(value, (dict, bool)): # Handle array items schema filtered_items = _filter_boolean_schemas(value) if filtered_items is not None: result[key] = filtered_items elif key == "properties" and isinstance(value, dict): # Recursively filter property schemas filtered_props = {} for prop_name, prop_schema in value.items(): filtered_prop = _filter_boolean_schemas(prop_schema) if filtered_prop is not None: filtered_props[prop_name] = filtered_prop result[key] = filtered_props elif key in ("$defs", "definitions") and isinstance(value, dict): # Recursively filter definitions filtered_defs = {} for def_name, def_schema in value.items(): filtered_def = _filter_boolean_schemas(def_schema) if filtered_def is not None: filtered_defs[def_name] = filtered_def result[key] = filtered_defs else: result[key] = value return result def json_schema_to_pydantic_type( json_schema: t.Union[t.Dict[str, t.Any], bool], ) -> t.Union[t.Type, t.Optional[t.Any]]: """ Converts a JSON schema type to a Pydantic type. Uses json-schema-to-pydantic for complex schemas (anyOf, allOf, oneOf), falls back to simple type mapping for primitive types. :param json_schema: The JSON schema to convert (can be dict or boolean). :return: A Pydantic type. """ # Handle boolean schemas (JSON Schema draft-06+) if isinstance(json_schema, bool): if json_schema: return t.Any # true schema accepts any value else: return None # false schema - will be filtered out in union processing # Pre-filter boolean schemas from combiners filtered_schema = _filter_boolean_schemas(json_schema) if filtered_schema is None: return str # Fallback if all schemas were false # Handle simple primitive types without complex combiners if _is_simple_primitive(filtered_schema): return _convert_simple_type(filtered_schema) # Use library for complex schemas (anyOf, allOf, oneOf, nested objects) return _convert_with_library(filtered_schema) def _is_simple_primitive(schema: t.Dict[str, t.Any]) -> bool: """Check if schema is a simple primitive without combiners.""" has_combiners = any(k in schema for k in ("anyOf", "allOf", "oneOf")) has_properties = "properties" in schema schema_type = schema.get("type") return ( not has_combiners and not has_properties and schema_type in PYDANTIC_TYPE_TO_PYTHON_TYPE and schema_type not in CONTAINER_TYPE ) def _convert_simple_type(schema: t.Dict[str, t.Any]) -> t.Type[t.Any]: """Convert simple primitive types directly.""" type_ = schema.get("type", "string") return t.cast(t.Type[t.Any], PYDANTIC_TYPE_TO_PYTHON_TYPE.get(type_, str)) def _convert_with_library( schema: t.Dict[str, t.Any], ) -> t.Union[t.Type, t.Any]: """Use json-schema-to-pydantic for complex schema conversion.""" try: # Handle top-level combiner without type (e.g., {"anyOf": [...]}) if ( any(k in schema for k in ("anyOf", "allOf", "oneOf")) and "type" not in schema ): return _handle_toplevel_combiner(schema) # For object schemas, create model directly if schema.get("type") == "object": if "title" not in schema: schema = {**schema, "title": "GeneratedModel"} return create_model_from_schema( schema, allow_undefined_array_items=True, allow_undefined_type=True, ) # For array schemas if schema.get("type") == "array": items = schema.get("items") if items: item_type = json_schema_to_pydantic_type(items) return t.List[t.cast(t.Type, item_type)] # type: ignore return t.List # Fallback to simple type return _convert_simple_type(schema) except (SchemaError, CombinerError) as e: logger.debug(f"Library schema conversion failed: {e}, falling back to string") return str except Exception as e: logger.debug( f"Unexpected error in schema conversion: {e}, falling back to string" ) return str def _handle_toplevel_combiner( schema: t.Dict[str, t.Any], ) -> t.Union[t.Type, t.Any]: """ Handle top-level combiner schemas (anyOf, allOf, oneOf without "type"). The library can handle these directly - it returns the appropriate type. """ try: # Try direct conversion - library handles anyOf/oneOf/allOf at top level result = create_model_from_schema( schema, allow_undefined_array_items=True, allow_undefined_type=True, ) if result is type(None): return t.Optional[t.Any] # If result is a type (like a Union or Optional), return it directly # If result is a model class, return it return result except (SchemaError, CombinerError): pass except Exception: pass # Fallback: manually build union type for anyOf/oneOf if "anyOf" in schema or "oneOf" in schema: options = schema.get("anyOf", schema.get("oneOf", [])) return _build_union_from_options(options) # Fallback: use first option for allOf if "allOf" in schema and schema["allOf"]: return json_schema_to_pydantic_type(schema["allOf"][0]) return t.Any def _build_union_from_options(options: t.List[t.Dict[str, t.Any]]) -> t.Type: """Build a Union type from a list of schema options.""" pydantic_types = [] null_type = PYDANTIC_TYPE_TO_PYTHON_TYPE.get("null") has_null = False for option in options: ptype = json_schema_to_pydantic_type(option) if ptype is None: continue if ptype == null_type or ptype is type(None): has_null = True continue pydantic_types.append(ptype) if len(pydantic_types) == 0: return t.Optional[t.Any] if has_null else str # type: ignore if len(pydantic_types) == 1: base_type = pydantic_types[0] if has_null: return t.Optional[t.cast(t.Type, base_type)] # type: ignore return base_type # Build union cast_types = [t.cast(t.Type, ptype) for ptype in pydantic_types] union_type = reduce(lambda a, b: t.Union[a, b], cast_types) # type: ignore if has_null: return t.Optional[union_type] # type: ignore return union_type