|
1 | | -from typing import Any, Dict, List, Optional, Set, Type, TypeVar |
| 1 | +from typing import Annotated, Any, Dict, List, Optional, Set, Type, TypeVar |
2 | 2 |
|
3 | | -from pydantic import BaseModel, Field, create_model, ConfigDict |
| 3 | +from pydantic import BaseModel, Field, RootModel, create_model, ConfigDict |
4 | 4 |
|
5 | 5 | from .builders import ConstraintBuilder |
6 | 6 | from .handlers import CombinerHandler |
@@ -29,7 +29,7 @@ class PydanticModelBuilder(IModelBuilder[T]): |
29 | 29 | "type", "title", "description", "properties", "required", "additionalProperties", |
30 | 30 | "patternProperties", "dependencies", "propertyNames", "if", "then", "else", |
31 | 31 | "allOf", "anyOf", "oneOf", "not", "$ref", "$defs", "definitions", "$schema", |
32 | | - "$id", "$comment" |
| 32 | + "$id", "$comment", "items", "minItems", "maxItems", "uniqueItems" |
33 | 33 | } |
34 | 34 |
|
35 | 35 | def __init__(self, base_model_type: Type[T] = BaseModel): |
@@ -106,6 +106,12 @@ def create_pydantic_model( |
106 | 106 | schema["oneOf"], root_schema, allow_undefined_array_items, allow_undefined_type |
107 | 107 | ) |
108 | 108 |
|
| 109 | + # Handle top-level arrays |
| 110 | + if schema.get("type") == "array": |
| 111 | + return self._create_array_root_model( |
| 112 | + schema, root_schema, allow_undefined_array_items, original_ref |
| 113 | + ) |
| 114 | + |
109 | 115 | # Get model properties |
110 | 116 | # If this schema is referenced, use the ref name as the title if no title is provided |
111 | 117 | if original_ref and "title" not in schema: |
@@ -165,6 +171,106 @@ class DynamicBase(self.base_model_type): |
165 | 171 |
|
166 | 172 | return model |
167 | 173 |
|
| 174 | + def _create_array_root_model( |
| 175 | + self, |
| 176 | + schema: Dict[str, Any], |
| 177 | + root_schema: Dict[str, Any], |
| 178 | + allow_undefined_array_items: bool, |
| 179 | + original_ref: Optional[str], |
| 180 | + ) -> Type[T]: |
| 181 | + """ |
| 182 | + Creates a RootModel for top-level array schemas. |
| 183 | + |
| 184 | + Args: |
| 185 | + schema: The array schema definition |
| 186 | + root_schema: The root schema containing definitions |
| 187 | + allow_undefined_array_items: Whether to allow arrays without items |
| 188 | + original_ref: The original reference if this was a $ref |
| 189 | + |
| 190 | + Returns: |
| 191 | + A RootModel class that validates arrays |
| 192 | + """ |
| 193 | + # Get the title for the model |
| 194 | + if original_ref and "title" not in schema: |
| 195 | + ref_parts = original_ref.split("/") |
| 196 | + title = ref_parts[-1] if ref_parts else "DynamicModel" |
| 197 | + else: |
| 198 | + title = schema.get("title", "DynamicModel") |
| 199 | + |
| 200 | + # Get description |
| 201 | + description = schema.get("description") |
| 202 | + |
| 203 | + # Resolve the item type |
| 204 | + items_schema = schema.get("items") |
| 205 | + if not items_schema: |
| 206 | + if allow_undefined_array_items: |
| 207 | + item_type = Any |
| 208 | + else: |
| 209 | + from .exceptions import TypeError |
| 210 | + raise TypeError("Array type must specify 'items' schema") |
| 211 | + else: |
| 212 | + item_type = self._get_field_type( |
| 213 | + items_schema, root_schema, allow_undefined_array_items |
| 214 | + ) |
| 215 | + |
| 216 | + # Determine if we need to use Set or List |
| 217 | + if schema.get("uniqueItems", False): |
| 218 | + array_type = Set[item_type] |
| 219 | + else: |
| 220 | + array_type = List[item_type] |
| 221 | + |
| 222 | + # Build constraints for the array |
| 223 | + constraints = self.constraint_builder.build_constraints(schema) |
| 224 | + |
| 225 | + # Extract model-level json_schema_extra (non-standard properties) |
| 226 | + model_extra = { |
| 227 | + key: value for key, value in schema.items() |
| 228 | + if key not in self.STANDARD_MODEL_PROPERTIES |
| 229 | + } |
| 230 | + |
| 231 | + # Create the RootModel class dynamically |
| 232 | + # RootModel requires the type to be specified as a generic parameter |
| 233 | + # We create a class that properly inherits from RootModel[array_type] |
| 234 | + |
| 235 | + # Build the class namespace |
| 236 | + namespace = {} |
| 237 | + if description: |
| 238 | + namespace["__doc__"] = description |
| 239 | + |
| 240 | + # Add model_config if we have extra properties |
| 241 | + if model_extra: |
| 242 | + namespace["model_config"] = ConfigDict(json_schema_extra=model_extra) |
| 243 | + |
| 244 | + # Apply constraints to the root field using Annotated |
| 245 | + if constraints: |
| 246 | + # Use Annotated to add Field constraints to the array type |
| 247 | + root_type = Annotated[array_type, Field(**constraints)] |
| 248 | + else: |
| 249 | + root_type = array_type |
| 250 | + namespace["__annotations__"] = {"root": root_type} |
| 251 | + |
| 252 | + # Create the RootModel subclass |
| 253 | + model = type( |
| 254 | + title, |
| 255 | + (RootModel[array_type],), |
| 256 | + namespace |
| 257 | + ) |
| 258 | + |
| 259 | + # Cache the model if it was referenced |
| 260 | + if original_ref: |
| 261 | + self._model_cache[original_ref] = model |
| 262 | + # Mark model for rebuild if needed |
| 263 | + self._models_to_rebuild.add(model) |
| 264 | + |
| 265 | + # Rebuild models if this is the top-level call |
| 266 | + if not self._building_models and self._models_to_rebuild: |
| 267 | + namespace = {m.__name__: m for m in self._models_to_rebuild} |
| 268 | + for m in self._models_to_rebuild: |
| 269 | + m.model_rebuild(_types_namespace=namespace) |
| 270 | + self._models_to_rebuild.clear() |
| 271 | + |
| 272 | + return model |
| 273 | + |
168 | 274 | def _get_field_type( |
169 | 275 | self, |
170 | 276 | field_schema: Dict[str, Any], |
|
0 commit comments