Recursively processes a schema property to determine its Python type hint and Pydantic Field definition. Returns: A tuple containing (python_type_hint, pydantic_field). The pydantic_field contains default value and description.
(
_model_cache: Dict[str, Type],
prop_schema: Dict[str, Any],
model_name_prefix: str,
prop_name: str,
is_required: bool,
schema_defs: Optional[Dict] = None,
)
| 89 | |
| 90 | |
| 91 | def _process_schema_property( |
| 92 | _model_cache: Dict[str, Type], |
| 93 | prop_schema: Dict[str, Any], |
| 94 | model_name_prefix: str, |
| 95 | prop_name: str, |
| 96 | is_required: bool, |
| 97 | schema_defs: Optional[Dict] = None, |
| 98 | ) -> tuple[Union[Type, List, ForwardRef, Any], FieldInfo]: |
| 99 | """ |
| 100 | Recursively processes a schema property to determine its Python type hint |
| 101 | and Pydantic Field definition. |
| 102 | |
| 103 | Returns: |
| 104 | A tuple containing (python_type_hint, pydantic_field). |
| 105 | The pydantic_field contains default value and description. |
| 106 | """ |
| 107 | if "$ref" in prop_schema: |
| 108 | ref = prop_schema["$ref"] |
| 109 | if ref.startswith("#/properties/"): |
| 110 | # Remove common prefix in pathes. |
| 111 | prefix_path = model_name_prefix.split("_form_model_")[-1] |
| 112 | ref_path = ref.split("#/properties/")[-1] |
| 113 | # Translate $ref path to model_name_prefix style. |
| 114 | ref_path = ref_path.replace("/properties/", "_model_") |
| 115 | ref_path = ref_path.replace("/items", "_item") |
| 116 | # If $ref path is a prefix substring of model_name_prefix path, |
| 117 | # there exists a circular reference. |
| 118 | # The loop should be broke with a return to avoid exception. |
| 119 | if prefix_path.startswith(ref_path): |
| 120 | # TODO: Find the exact type hint for the $ref. |
| 121 | return Any, Field(default=None, description="") |
| 122 | ref = ref.split("/")[-1] |
| 123 | assert ref in schema_defs, "Custom field not found" |
| 124 | prop_schema = schema_defs[ref] |
| 125 | |
| 126 | prop_type = prop_schema.get("type") |
| 127 | prop_desc = prop_schema.get("description", "") |
| 128 | |
| 129 | default_value = ... if is_required else prop_schema.get("default", None) |
| 130 | pydantic_field = Field(default=default_value, description=prop_desc) |
| 131 | |
| 132 | # Handle the case where prop_type is missing but 'anyOf' key exists |
| 133 | # In this case, use data type from 'anyOf' to determine the type hint |
| 134 | if "anyOf" in prop_schema: |
| 135 | type_hints = [] |
| 136 | for i, schema_option in enumerate(prop_schema["anyOf"]): |
| 137 | type_hint, _ = _process_schema_property( |
| 138 | _model_cache, |
| 139 | schema_option, |
| 140 | f"{model_name_prefix}_{prop_name}", |
| 141 | f"choice_{i}", |
| 142 | False, |
| 143 | schema_defs=schema_defs, |
| 144 | ) |
| 145 | type_hints.append(type_hint) |
| 146 | return Union[tuple(type_hints)], pydantic_field |
| 147 | |
| 148 | # Handle the case where prop_type is a list of types, e.g. ['string', 'number'] |