From abe67df95eab51e770b31a458a9bbce66869870e Mon Sep 17 00:00:00 2001 From: Ishaan Jaffer Date: Wed, 22 Oct 2025 10:04:52 -0700 Subject: [PATCH] refactor large func --- .../proxy/guardrails/guardrail_endpoints.py | 199 ++++++++++-------- 1 file changed, 114 insertions(+), 85 deletions(-) diff --git a/litellm/proxy/guardrails/guardrail_endpoints.py b/litellm/proxy/guardrails/guardrail_endpoints.py index c1756500c6..29da0ec40e 100644 --- a/litellm/proxy/guardrails/guardrail_endpoints.py +++ b/litellm/proxy/guardrails/guardrail_endpoints.py @@ -809,6 +809,105 @@ def _get_list_element_options(field_annotation: Any) -> Optional[List[str]]: return None +def _should_skip_optional_params(field_name: str, field_annotation: Any) -> bool: + """Check if optional_params field should be skipped (not meaningfully overridden).""" + if field_name != "optional_params": + return False + + if field_annotation is None: + return True + + # Check if the annotation is still a generic TypeVar (not specialized) + if isinstance(field_annotation, TypeVar) or ( + hasattr(field_annotation, "__origin__") + and field_annotation.__origin__ is TypeVar + ): + return True + + # Also skip if it's a generic type that wasn't specialized + if hasattr(field_annotation, "__name__") and field_annotation.__name__ in ( + "T", + "TypeVar", + ): + return True + + # Handle Optional[T] where T is still a TypeVar + if hasattr(field_annotation, "__args__"): + non_none_args = [arg for arg in field_annotation.__args__ if arg is not type(None)] + if non_none_args and isinstance(non_none_args[0], TypeVar): + return True + + return False + + +def _unwrap_optional_type(field_annotation: Any) -> Any: + """Unwrap Optional types to get the actual type.""" + if ( + hasattr(field_annotation, "__origin__") + and field_annotation.__origin__ is Union + and hasattr(field_annotation, "__args__") + ): + # For Optional[BaseModel], get the non-None type + args = field_annotation.__args__ + non_none_args = [arg for arg in args if arg is not type(None)] + if non_none_args: + return non_none_args[0] + return field_annotation + + +def _build_field_dict( + field: Any, + field_annotation: Any, + description: str, + required: bool, +) -> Dict[str, Any]: + """Build field dictionary for non-nested fields.""" + # Determine the field type from annotation + field_type = _get_field_type_from_annotation(field_annotation) + + # Check for custom UI type override + field_json_schema_extra = getattr(field, "json_schema_extra", {}) + if field_json_schema_extra and "ui_type" in field_json_schema_extra: + field_type = field_json_schema_extra["ui_type"].value + elif field_json_schema_extra and "type" in field_json_schema_extra: + field_type = field_json_schema_extra["type"] + + # Add the field to the dictionary + field_dict = { + "description": description, + "required": required, + "type": field_type, + } + + # Extract options from type annotations + if field_type == "dict": + # For Dict[Literal[...], T] types, extract key options + dict_key_options = _get_dict_key_options(field_annotation) + if dict_key_options: + field_dict["dict_key_options"] = dict_key_options + + # Extract value type for the dict values + dict_value_type = _get_dict_value_type(field_annotation) + field_dict["dict_value_type"] = dict_value_type + + elif field_type == "array": + # For List[Literal[...]] types, extract element options + list_element_options = _get_list_element_options(field_annotation) + if list_element_options: + field_dict["options"] = list_element_options + field_dict["type"] = "multiselect" + + # Add options if they exist in json_schema_extra (this takes precedence) + if field_json_schema_extra and "options" in field_json_schema_extra: + field_dict["options"] = field_json_schema_extra["options"] + + # Add default value if it exists + if field.default is not None and field.default is not ...: + field_dict["default_value"] = field.default + + return field_dict + + def _extract_fields_recursive( model: Type[BaseModel], depth: int = 0, @@ -823,53 +922,21 @@ def _extract_fields_recursive( fields = {} for field_name, field in model.model_fields.items(): - # Skip optional_params if it's not meaningfully overridden - if field_name == "optional_params": - field_annotation = field.annotation - if field_annotation is None: - continue - # Check if the annotation is still a generic TypeVar (not specialized) - if isinstance(field_annotation, TypeVar) or ( - hasattr(field_annotation, "__origin__") - and field_annotation.__origin__ is TypeVar - ): - # Skip this field as it's not meaningfully overridden - continue - # Also skip if it's a generic type that wasn't specialized - if hasattr(field_annotation, "__name__") and field_annotation.__name__ in ( - "T", - "TypeVar", - ): - continue - # Handle Optional[T] where T is still a TypeVar - if hasattr(field_annotation, "__args__"): - non_none_args = [arg for arg in field_annotation.__args__ if arg is not type(None)] - if non_none_args and isinstance(non_none_args[0], TypeVar): - continue - - # Get field metadata - description = field.description or field_name - - # Check if this field is required - required = field.is_required() - - # Check if the field annotation is a BaseModel subclass field_annotation = field.annotation + + # Skip optional_params if it's not meaningfully overridden + if _should_skip_optional_params(field_name=field_name, field_annotation=field_annotation): + continue # Handle Optional types and get the actual type if field_annotation is None: continue - if ( - hasattr(field_annotation, "__origin__") - and field_annotation.__origin__ is Union - and hasattr(field_annotation, "__args__") - ): - # For Optional[BaseModel], get the non-None type - args = field_annotation.__args__ - non_none_args = [arg for arg in args if arg is not type(None)] - if non_none_args: - field_annotation = non_none_args[0] + field_annotation = _unwrap_optional_type(field_annotation=field_annotation) + + # Get field metadata + description = field.description or field_name + required = field.is_required() # Check if this is a BaseModel subclass is_basemodel_subclass = ( @@ -890,50 +957,12 @@ def _extract_fields_recursive( "fields": nested_fields, } else: - # Determine the field type from annotation - field_type = _get_field_type_from_annotation(field_annotation) - - # Check for custom UI type override - field_json_schema_extra = getattr(field, "json_schema_extra", {}) - if field_json_schema_extra and "ui_type" in field_json_schema_extra: - field_type = field_json_schema_extra["ui_type"].value - elif field_json_schema_extra and "type" in field_json_schema_extra: - field_type = field_json_schema_extra["type"] - - # Add the field to the dictionary - field_dict = { - "description": description, - "required": required, - "type": field_type, - } - - # Extract options from type annotations - if field_type == "dict": - # For Dict[Literal[...], T] types, extract key options - dict_key_options = _get_dict_key_options(field_annotation) - if dict_key_options: - field_dict["dict_key_options"] = dict_key_options - - # Extract value type for the dict values - dict_value_type = _get_dict_value_type(field_annotation) - field_dict["dict_value_type"] = dict_value_type - - elif field_type == "array": - # For List[Literal[...]] types, extract element options - list_element_options = _get_list_element_options(field_annotation) - if list_element_options: - field_dict["options"] = list_element_options - field_dict["type"] = "multiselect" - - # Add options if they exist in json_schema_extra (this takes precedence) - if field_json_schema_extra and "options" in field_json_schema_extra: - field_dict["options"] = field_json_schema_extra["options"] - - # Add default value if it exists - if field.default is not None and field.default is not ...: - field_dict["default_value"] = field.default - - fields[field_name] = field_dict + fields[field_name] = _build_field_dict( + field=field, + field_annotation=field_annotation, + description=description, + required=required, + ) return fields