|
1 | 1 | import re |
2 | 2 | from collections import UserString |
| 3 | +from collections.abc import Iterable |
3 | 4 | from collections.abc import Iterator |
4 | 5 | from inspect import isclass |
5 | 6 | from typing import TYPE_CHECKING |
@@ -41,6 +42,19 @@ def _is_in_schema(path_lower: str, schema: str) -> bool: |
41 | 42 | return path_lower == schema_lower or path_lower.startswith(f"{schema_lower}:") |
42 | 43 |
|
43 | 44 |
|
| 45 | +def _designating_schema(path_lower: str, schemas: Iterable[str]) -> str | None: |
| 46 | + """Return the schema a qualified path is expressed in, if any. |
| 47 | +
|
| 48 | + The longest match wins: the URN of an extension may extend the URN of the |
| 49 | + resource it extends, and then both match while only the longer one |
| 50 | + designates the model the attribute lives on. |
| 51 | + """ |
| 52 | + matching = [ |
| 53 | + schema for schema in schemas if schema and _is_in_schema(path_lower, schema) |
| 54 | + ] |
| 55 | + return max(matching, key=len) if matching else None |
| 56 | + |
| 57 | + |
44 | 58 | def _to_comparable(value: Any) -> Any: |
45 | 59 | """Convert a value to a comparable form (dict for BaseModel).""" |
46 | 60 | return value.model_dump() if isinstance(value, BaseModel) else value |
@@ -334,23 +348,20 @@ def _resolve_model(self) -> tuple[type[BaseModel], str | None] | None: |
334 | 348 | if ":" in self and isclass(model) and issubclass(model, Resource | Extension): |
335 | 349 | path_lower = str(self).lower() |
336 | 350 |
|
337 | | - if model.__schema__ and path_lower == model.__schema__.lower(): |
| 351 | + extension_models = ( |
| 352 | + model.get_extension_models() if issubclass(model, Resource) else {} |
| 353 | + ) |
| 354 | + schema = _designating_schema( |
| 355 | + path_lower, [model.__schema__ or "", *extension_models] |
| 356 | + ) |
| 357 | + |
| 358 | + if schema is None: |
| 359 | + return None |
| 360 | + |
| 361 | + model = extension_models.get(schema, model) |
| 362 | + if path_lower == schema.lower(): |
338 | 363 | return model, None |
339 | | - elif model.__schema__ and _is_in_schema(path_lower, model.__schema__): |
340 | | - attr_path = str(self)[len(model.__schema__) :].lstrip(":") |
341 | | - elif issubclass(model, Resource): |
342 | | - for ( |
343 | | - extension_schema, |
344 | | - extension_model, |
345 | | - ) in model.get_extension_models().items(): |
346 | | - schema_lower = extension_schema.lower() |
347 | | - if path_lower == schema_lower: |
348 | | - return extension_model, None |
349 | | - elif _is_in_schema(path_lower, schema_lower): |
350 | | - model = extension_model |
351 | | - break |
352 | | - else: |
353 | | - return None |
| 364 | + attr_path = str(self)[len(schema) :].lstrip(":") |
354 | 365 |
|
355 | 366 | if not attr_path: |
356 | 367 | return model, None |
@@ -397,34 +408,36 @@ def _resolve_instance( |
397 | 408 | if ":" not in path_str: |
398 | 409 | return _Resolution(resource, path_str) |
399 | 410 |
|
400 | | - model_schema = getattr(type(resource), "__schema__", "") or "" |
401 | 411 | path_lower = path_str.lower() |
| 412 | + model_schema = "" |
| 413 | + if isinstance(resource, Resource | Extension): |
| 414 | + model_schema = getattr(type(resource), "__schema__", "") or "" |
| 415 | + extension_models = ( |
| 416 | + resource.get_extension_models() if isinstance(resource, Resource) else {} |
| 417 | + ) |
| 418 | + |
| 419 | + schema = _designating_schema(path_lower, [model_schema, *extension_models]) |
| 420 | + if schema is None: |
| 421 | + if isinstance(resource, Resource): |
| 422 | + raise InvalidPathException(path=path_str) |
| 423 | + return None |
402 | 424 |
|
403 | | - if isinstance(resource, Resource | Extension) and _is_in_schema( |
404 | | - path_lower, model_schema |
405 | | - ): |
406 | | - is_explicit = path_lower == model_schema.lower() |
407 | | - normalized = path_str[len(model_schema) :].lstrip(":") |
408 | | - return _Resolution(resource, normalized, is_explicit) |
409 | | - |
410 | | - if isinstance(resource, Resource): |
411 | | - for ext_schema, ext_model in resource.get_extension_models().items(): |
412 | | - ext_schema_lower = ext_schema.lower() |
413 | | - if path_lower == ext_schema_lower: |
414 | | - return _Resolution(resource, ext_model.__name__) |
415 | | - if _is_in_schema(path_lower, ext_schema_lower): |
416 | | - sub_path = path_str[len(ext_schema) :].lstrip(":") |
417 | | - ext_obj = getattr(resource, ext_model.__name__) |
418 | | - if create and ext_obj is None: |
419 | | - ext_obj = ext_model() |
420 | | - setattr(resource, ext_model.__name__, ext_obj) |
421 | | - if ext_obj is None: |
422 | | - return None |
423 | | - return _Resolution(ext_obj, sub_path) |
| 425 | + sub_path = path_str[len(schema) :].lstrip(":") |
| 426 | + ext_model = extension_models.get(schema) |
424 | 427 |
|
425 | | - raise InvalidPathException(path=str(self)) |
| 428 | + if ext_model is None: |
| 429 | + return _Resolution(resource, sub_path, path_lower == schema.lower()) |
426 | 430 |
|
427 | | - return None |
| 431 | + if path_lower == schema.lower(): |
| 432 | + return _Resolution(resource, ext_model.__name__) |
| 433 | + |
| 434 | + ext_obj = getattr(resource, ext_model.__name__) |
| 435 | + if create and ext_obj is None: |
| 436 | + ext_obj = ext_model() |
| 437 | + setattr(resource, ext_model.__name__, ext_obj) |
| 438 | + if ext_obj is None: |
| 439 | + return None |
| 440 | + return _Resolution(ext_obj, sub_path) |
428 | 441 |
|
429 | 442 | def _walk_to_target( |
430 | 443 | self, obj: BaseModel, path_str: str, *, create: bool = False |
|
0 commit comments