import re
from collections import UserString
from collections.abc import Iterator
from inspect import isclass
from typing import TYPE_CHECKING
from typing import Any
from typing import Generic
from typing import NamedTuple
from typing import TypeVar
from pydantic import GetCoreSchemaHandler
from pydantic import GetJsonSchemaHandler
from pydantic.json_schema import JsonSchemaValue
from pydantic_core import core_schema
from .base import BaseModel
from .utils import _find_field_name
from .utils import _to_camel
if TYPE_CHECKING:
from .annotations import CaseExact
from .annotations import Mutability
from .annotations import Required
from .annotations import Returned
from .annotations import Uniqueness
from .resources.resource import Resource
from .exceptions import InvalidPathException
from .exceptions import PathNotFoundException
ResourceT = TypeVar("ResourceT", bound="Resource[Any]")
_VALID_PATH_PATTERN = re.compile(r'^[a-zA-Z][a-zA-Z0-9._:\-\[\]"=\s]*$')
_PATH_CACHE: dict[tuple[type, type], type] = {}
def _to_comparable(value: Any) -> Any:
"""Convert a value to a comparable form (dict for BaseModel)."""
return value.model_dump() if isinstance(value, BaseModel) else value
def _values_match(value1: Any, value2: Any) -> bool:
"""Check if two values match, handling BaseModel comparison."""
return bool(_to_comparable(value1) == _to_comparable(value2))
def _value_in_list(current_list: list[Any], new_value: Any) -> bool:
"""Check if a value exists in a list, handling BaseModel comparison."""
return any(_values_match(item, new_value) for item in current_list)
def _require_field(model: type[BaseModel], name: str) -> str:
"""Find field name or raise PathNotFoundException."""
if (field_name := _find_field_name(model, name)) is None:
raise PathNotFoundException(path=name, field=name)
return field_name
class _Resolution(NamedTuple):
"""Result of instance path resolution."""
target: "BaseModel"
path_str: str
is_explicit_schema_path: bool = False
[docs]
class URN(str):
"""URN string type with validation."""
def __new__(cls, urn: str) -> "URN":
cls.check_syntax(urn)
return super().__new__(cls, urn)
@classmethod
def __get_pydantic_core_schema__(
cls,
_source: type[Any],
_handler: GetCoreSchemaHandler,
) -> core_schema.CoreSchema:
return core_schema.no_info_after_validator_function(
cls,
core_schema.str_schema(),
serialization=core_schema.plain_serializer_function_ser_schema(
str,
),
)
[docs]
@classmethod
def check_syntax(cls, path: str) -> None:
"""Validate URN-based path format.
:param path: The URN path to validate
:raises ValueError: If the URN format is invalid
"""
if not path.startswith("urn:"):
raise ValueError("The URN does not start with urn:")
urn_segments = path.split(":")
if len(urn_segments) < 3:
raise ValueError("URNs must have at least 3 parts")
[docs]
class Path(UserString, Generic[ResourceT]):
__scim_model__: type[BaseModel] | None = None
def __class_getitem__(cls, model: type[ResourceT]) -> type["Path[ResourceT]"]:
"""Create a Path class bound to a specific model type."""
if not isclass(model) or not hasattr(model, "model_fields"):
return super().__class_getitem__(model) # type: ignore[misc,no-any-return]
cache_key = (cls, model)
if cache_key in _PATH_CACHE:
return _PATH_CACHE[cache_key]
new_class = type(f"Path[{model.__name__}]", (cls,), {"__scim_model__": model})
_PATH_CACHE[cache_key] = new_class
return new_class
@classmethod
def __get_pydantic_core_schema__(
cls,
_source_type: type[Any],
_handler: GetCoreSchemaHandler,
) -> core_schema.CoreSchema:
def validate_path(value: Any) -> "Path[Any]":
if isinstance(value, Path):
return cls(str(value))
if isinstance(value, str):
return cls(value)
raise ValueError(f"Expected str or Path, got {type(value).__name__}")
return core_schema.no_info_plain_validator_function(
validate_path,
serialization=core_schema.plain_serializer_function_ser_schema(str),
)
@classmethod
def __get_pydantic_json_schema__(
cls,
_core_schema: core_schema.CoreSchema,
_handler: GetJsonSchemaHandler,
) -> JsonSchemaValue:
return {"type": "string"}
def __init__(self, path: "str | Path[Any]"):
if isinstance(path, Path):
path = str(path)
self.check_syntax(path)
self.data = path
[docs]
@classmethod
def check_syntax(cls, path: str) -> None:
"""Check if path syntax is valid according to RFC 7644 simplified rules.
An empty string is valid and represents the resource root.
:param path: The path to validate
:raises ValueError: If the path syntax is invalid
"""
if not path:
return
if path[0].isdigit():
raise ValueError("Paths cannot start with a digit")
if ".." in path:
raise ValueError("Paths cannot contain double dots")
if not _VALID_PATH_PATTERN.match(path):
raise ValueError("The path contains invalid characters")
if path.endswith(":"):
raise ValueError("Paths cannot end with a colon")
if ":" in path:
urn = path.rsplit(":", 1)[0]
try:
URN(urn.lower())
except ValueError as exc:
raise ValueError(f"The path is not a valid URN: {exc}") from exc
@property
def schema(self) -> str | None:
"""The schema URN portion of the path.
For paths like "urn:...:User:userName", returns "urn:...:User".
For simple paths like "userName", returns None.
"""
if ":" not in self.data:
return None
return self.data.rsplit(":", 1)[0]
@property
def attr(self) -> str:
"""The attribute portion of the path.
For paths like "urn:...:User:userName", returns "userName".
For simple paths like "userName", returns "userName".
For schema-only paths like "urn:...:User", returns "".
"""
if ":" not in self.data:
return self.data
return self.data.rsplit(":", 1)[1]
@property
def parts(self) -> tuple[str, ...]:
"""The attribute path segments split by '.'.
For "name.familyName", returns ("name", "familyName").
For "userName", returns ("userName",).
For "", returns ().
"""
attr = self.attr
if not attr:
return ()
return tuple(attr.split("."))
[docs]
def is_prefix_of(self, other: "str | Path[Any]") -> bool:
"""Check if this path is a prefix of another path.
A path is a prefix if the other path starts with this path
followed by a separator ("." or ":").
Examples::
Path("emails").is_prefix_of("emails.value") # True
Path("emails").is_prefix_of("emails") # False (equal, not prefix)
Path("urn:...:User").is_prefix_of("urn:...:User:name") # True
"""
other_str = str(other).lower()
self_str = self.data.lower()
if self_str == other_str:
return False
return other_str.startswith(f"{self_str}.") or other_str.startswith(
f"{self_str}:"
)
[docs]
def has_prefix(self, prefix: "str | Path[Any]") -> bool:
"""Check if this path has the given prefix.
Examples::
Path("emails.value").has_prefix("emails") # True
Path("emails").has_prefix("emails") # False (equal, not prefix)
Path("urn:...:User:name").has_prefix("urn:...:User") # True
"""
prefix_path = prefix if isinstance(prefix, Path) else Path(str(prefix))
return prefix_path.is_prefix_of(self)
@property
def model(self) -> type[BaseModel] | None:
"""The target model type for this path.
Requires the Path to be bound to a model type via ``Path[Model]``.
Returns None if the path is unbound or invalid.
For "name.familyName" on Path[User], returns Name.
For "userName" on Path[User], returns User.
"""
if (result := self._resolve_model()) is None:
return None
return result[0]
@property
def field_name(self) -> str | None:
"""The Python attribute name (snake_case) for this path.
Requires the Path to be bound to a model type via ``Path[Model]``.
Returns None if the path is unbound or invalid.
For "name.familyName" on Path[User], returns "family_name".
For "userName" on Path[User], returns "user_name".
"""
if (result := self._resolve_model()) is None:
return None
return result[1]
@property
def field_type(self) -> type | None:
"""The Python type of the field this path points to.
Requires the Path to be bound to a model type via ``Path[Model]``.
Returns None if the path is unbound, invalid, or points to a schema-only path.
For "userName" on Path[User], returns str.
For "name" on Path[User], returns Name.
For "emails" on Path[User], returns Email.
"""
if self.model is None or self.field_name is None:
return None
return self.model.get_field_root_type(self.field_name)
@property
def is_multivalued(self) -> bool | None:
"""Whether this path points to a multi-valued attribute.
Requires the Path to be bound to a model type via ``Path[Model]``.
Returns None if the path is unbound, invalid, or points to a schema-only path.
For "emails" on Path[User], returns True.
For "userName" on Path[User], returns False.
"""
if self.model is None or self.field_name is None:
return None
return self.model.get_field_multiplicity(self.field_name)
[docs]
def get_annotation(self, annotation_type: type) -> Any:
"""Get annotation value for this path's field.
Requires the Path to be bound to a model type via ``Path[Model]``.
Returns None if the path is unbound, invalid, or points to a schema-only path.
:param annotation_type: The annotation class (e.g., Required, Mutability).
:returns: The annotation value or None.
For "userName" on Path[User] with Required, returns Required.true.
"""
if self.model is None or self.field_name is None:
return None
return self.model.get_field_annotation(self.field_name, annotation_type)
@property
def urn(self) -> str | None:
"""The fully qualified URN for this path.
Requires the Path to be bound to a model type via ``Path[Model]``.
Returns None if the path is unbound or invalid.
For "userName" on Path[User], returns
"urn:ietf:params:scim:schemas:core:2.0:User:userName".
"""
from .resources.resource import Resource
if self.__scim_model__ is None or self.model is None:
return None
schema = self.schema
if not schema and issubclass(self.__scim_model__, Resource):
schema = self.__scim_model__.__schema__
if not self.attr:
return schema if schema else None
return f"{schema}:{self.attr}" if schema else self.attr
def _resolve_model(self) -> tuple[type[BaseModel], str | None] | None:
"""Resolve the path against the bound model type."""
from .resources.resource import Extension
from .resources.resource import Resource
model = self.__scim_model__
if model is None:
return None
attr_path = self.attr
if ":" in self and isclass(model) and issubclass(model, Resource | Extension):
path_lower = str(self).lower()
if model.__schema__ and path_lower == model.__schema__.lower():
return model, None
elif model.__schema__ and path_lower.startswith(model.__schema__.lower()):
attr_path = str(self)[len(model.__schema__) :].lstrip(":")
elif issubclass(model, Resource):
for (
extension_schema,
extension_model,
) in model.get_extension_models().items():
schema_lower = extension_schema.lower()
if path_lower == schema_lower:
return extension_model, None
elif path_lower.startswith(schema_lower):
model = extension_model
break
else:
return None
if not attr_path:
return model, None
if "." in attr_path:
parts = attr_path.split(".")
current_model = model
for part in parts[:-1]:
if (field_name := _find_field_name(current_model, part)) is None:
return None
field_type = current_model.get_field_root_type(field_name)
if (
field_type is None
or not isclass(field_type)
or not issubclass(field_type, BaseModel)
):
return None
current_model = field_type
if (field_name := _find_field_name(current_model, parts[-1])) is None:
return None
return current_model, field_name
if (field_name := _find_field_name(model, attr_path)) is None:
return None
return model, field_name
def _resolve_instance(
self, resource: BaseModel, *, create: bool = False
) -> _Resolution | None:
"""Resolve the target object and remaining path.
:param resource: The resource to resolve against.
:param create: If True, create extension instance if it doesn't exist.
:returns: Resolution with target object and path, or None if target doesn't exist.
:raises InvalidPathException: If the path references an unknown extension.
"""
from .resources.resource import Extension
from .resources.resource import Resource
path_str = str(self)
if ":" not in path_str:
return _Resolution(resource, path_str)
model_schema = getattr(type(resource), "__schema__", "") or ""
path_lower = path_str.lower()
if isinstance(resource, Resource | Extension) and path_lower.startswith(
model_schema.lower()
):
is_explicit = path_lower == model_schema.lower()
normalized = path_str[len(model_schema) :].lstrip(":")
return _Resolution(resource, normalized, is_explicit)
if isinstance(resource, Resource):
for ext_schema, ext_model in resource.get_extension_models().items():
ext_schema_lower = ext_schema.lower()
if path_lower == ext_schema_lower:
return _Resolution(resource, ext_model.__name__)
if path_lower.startswith(ext_schema_lower):
sub_path = path_str[len(ext_schema) :].lstrip(":")
ext_obj = getattr(resource, ext_model.__name__)
if create and ext_obj is None:
ext_obj = ext_model()
setattr(resource, ext_model.__name__, ext_obj)
if ext_obj is None:
return None
return _Resolution(ext_obj, sub_path)
raise InvalidPathException(path=str(self))
return None
def _walk_to_target(
self, obj: BaseModel, path_str: str
) -> tuple[BaseModel, str] | None:
"""Navigate to the target object and field.
:returns: (target_obj, field_name) or None if an intermediate is None.
"""
if "." not in path_str:
return obj, _require_field(type(obj), path_str)
parts = path_str.split(".")
current_obj = obj
for part in parts[:-1]:
field_name = _require_field(type(current_obj), part)
if (current_obj := getattr(current_obj, field_name)) is None:
return None
return current_obj, _require_field(type(current_obj), parts[-1])
def _get(self, resource: ResourceT) -> Any:
"""Get the value at this path from a resource."""
if (resolution := self._resolve_instance(resource)) is None:
return None
if not resolution.path_str:
return resolution.target
if (
result := self._walk_to_target(resolution.target, resolution.path_str)
) is None:
return None
obj, field_name = result
return getattr(obj, field_name)
[docs]
def get(self, resource: ResourceT, *, strict: bool = True) -> Any:
"""Get the value at this path from a resource.
:param resource: The resource to get the value from.
:param strict: If True, raise exceptions for invalid paths.
:returns: The value at this path, or None if the value is absent.
:raises PathNotFoundException: If strict and the path references a non-existent field.
:raises InvalidPathException: If strict and the path references an unknown extension.
"""
try:
return self._get(resource)
except InvalidPathException:
if strict:
raise
return None
def _set(self, resource: ResourceT, value: Any, *, is_add: bool = False) -> bool:
"""Set a value at this path on a resource."""
if (resolution := self._resolve_instance(resource, create=True)) is None:
return False
obj = resolution.target
path_str = resolution.path_str
is_explicit_schema_path = resolution.is_explicit_schema_path
if not path_str:
if not isinstance(value, dict):
if is_explicit_schema_path:
raise InvalidPathException(path=str(self))
return False
filtered_value = {
k: v
for k, v in value.items()
if _find_field_name(type(obj), k) is not None
}
if not filtered_value:
return False
old_data = obj.model_dump()
updated_data = {**old_data, **filtered_value}
if updated_data == old_data:
return False
updated_obj = type(obj).model_validate(updated_data)
obj.__dict__.update(updated_obj.__dict__)
return True
if "." not in path_str:
field_name = _require_field(type(obj), path_str)
return self._set_field_value(obj, field_name, value, is_add)
parts = path_str.split(".")
current_obj = obj
for part in parts[:-1]:
field_name = _require_field(type(current_obj), part)
if (sub_obj := getattr(current_obj, field_name)) is None:
field_type = type(current_obj).get_field_root_type(field_name)
if field_type is None or field_type is Any or not isclass(field_type):
return False
sub_obj = field_type()
setattr(current_obj, field_name, sub_obj)
elif isinstance(sub_obj, list):
return False
current_obj = sub_obj
field_name = _require_field(type(current_obj), parts[-1])
return self._set_field_value(current_obj, field_name, value, is_add)
[docs]
def set(
self,
resource: ResourceT,
value: Any,
*,
is_add: bool = False,
strict: bool = True,
) -> bool:
"""Set a value at this path on a resource.
:param resource: The resource to set the value on.
:param value: The value to set.
:param is_add: If True and the target is multi-valued, append to the
list instead of replacing. Duplicates are not added.
:param strict: If True, raise exceptions for invalid paths.
:returns: True if the value was set/added, False if unchanged.
:raises InvalidPathException: If strict and the path does not exist or is invalid.
"""
try:
return self._set(resource, value, is_add=is_add)
except InvalidPathException:
if strict:
raise
return False
@staticmethod
def _set_field_value(
obj: BaseModel, field_name: str, value: Any, is_add: bool
) -> bool:
"""Set or add a value to a field."""
is_multivalued = obj.get_field_multiplicity(field_name)
if is_add and is_multivalued:
current_list = getattr(obj, field_name) or []
if isinstance(value, list):
new_values = [v for v in value if not _value_in_list(current_list, v)]
if not new_values:
return False
setattr(obj, field_name, current_list + new_values)
else:
if _value_in_list(current_list, value):
return False
current_list.append(value)
setattr(obj, field_name, current_list)
return True
if is_multivalued and not isinstance(value, list) and value is not None:
value = [value]
old_value = getattr(obj, field_name)
if old_value == value:
return False
setattr(obj, field_name, value)
return True
def _delete(self, resource: ResourceT, value: Any | None = None) -> bool:
"""Delete a value at this path from a resource."""
if (resolution := self._resolve_instance(resource)) is None:
return False
if not resolution.path_str:
raise InvalidPathException(path=str(self))
if (
result := self._walk_to_target(resolution.target, resolution.path_str)
) is None:
return False
obj, field_name = result
if (current_value := getattr(obj, field_name)) is None:
return False
if value is not None:
if not isinstance(current_value, list):
return False
new_list = [
item for item in current_value if not _values_match(item, value)
]
if len(new_list) == len(current_value):
return False
setattr(obj, field_name, new_list if new_list else None)
return True
setattr(obj, field_name, None)
return True
[docs]
def delete(
self, resource: ResourceT, value: Any | None = None, *, strict: bool = True
) -> bool:
"""Delete a value at this path from a resource.
If value is None, the entire attribute is set to None.
If value is provided and the attribute is multi-valued,
only matching values are removed from the list.
:param resource: The resource to delete the value from.
:param value: Optional specific value to remove from a list.
:param strict: If True, raise exceptions for invalid paths.
:returns: True if a value was deleted, False if unchanged.
:raises InvalidPathException: If strict and the path does not exist or is invalid.
"""
try:
return self._delete(resource, value)
except InvalidPathException:
if strict:
raise
return False
[docs]
@classmethod
def iter_paths(
cls,
include_subattributes: bool = True,
include_extensions: bool = True,
required: "list[Required] | None" = None,
mutability: "list[Mutability] | None" = None,
uniqueness: "list[Uniqueness] | None" = None,
returned: "list[Returned] | None" = None,
case_exact: "list[CaseExact] | None" = None,
) -> "Iterator[Path[ResourceT]]":
"""Iterate over all paths for the bound model and its extensions.
Requires the Path to be bound to a model type via ``Path[Model]``.
:param include_subattributes: Whether to include sub-attribute paths.
:param include_extensions: Whether to include extension attributes.
:param required: Filter by Required annotation values (e.g., [Required.true]).
:param mutability: Filter by Mutability annotation values (e.g., [Mutability.read_write]).
:param uniqueness: Filter by Uniqueness annotation values (e.g., [Uniqueness.server]).
:param returned: Filter by Returned annotation values (e.g., [Returned.always]).
:param case_exact: Filter by CaseExact annotation values (e.g., [CaseExact.true]).
:yields: Path instances for each attribute matching the filters.
"""
from .annotations import CaseExact
from .annotations import Mutability
from .annotations import Required
from .annotations import Returned
from .annotations import Uniqueness
from .attributes import ComplexAttribute
from .resources.resource import Extension
from .resources.resource import Resource
model = cls.__scim_model__
if model is None:
raise TypeError("iter_paths requires a bound Path type: Path[Model]")
def matches_filters(target_model: type[BaseModel], field_name: str) -> bool:
if required is not None:
field_required = target_model.get_field_annotation(field_name, Required)
if field_required not in required:
return False
if mutability is not None:
field_mutability = target_model.get_field_annotation(
field_name, Mutability
)
if field_mutability not in mutability:
return False
if uniqueness is not None:
field_uniqueness = target_model.get_field_annotation(
field_name, Uniqueness
)
if field_uniqueness not in uniqueness:
return False
if returned is not None:
field_returned = target_model.get_field_annotation(field_name, Returned)
if field_returned not in returned:
return False
if case_exact is not None:
field_case_exact = target_model.get_field_annotation(
field_name, CaseExact
)
if field_case_exact not in case_exact:
return False
return True
def iter_model_paths(
target_model: type[Resource[Any] | Extension],
) -> "Iterator[Path[ResourceT]]":
for field_name in target_model.model_fields:
if field_name in ("meta", "id", "schemas"):
continue
if not matches_filters(target_model, field_name):
continue
field_type = target_model.get_field_root_type(field_name)
urn: str
if isclass(field_type) and issubclass(field_type, Extension):
if not include_extensions:
continue
urn = field_type.__schema__ or ""
elif isclass(target_model) and issubclass(target_model, Extension):
urn = target_model().get_attribute_urn(field_name)
else:
urn = _to_camel(field_name)
yield cls(urn)
is_complex = (
field_type is not None
and isclass(field_type)
and issubclass(field_type, ComplexAttribute)
)
if include_subattributes and is_complex:
for sub_field_name in field_type.model_fields: # type: ignore[union-attr]
if not matches_filters(field_type, sub_field_name): # type: ignore[arg-type]
continue
sub_urn = f"{urn}.{_to_camel(sub_field_name)}"
yield cls(sub_urn)
yield from iter_model_paths(model) # type: ignore[arg-type]
if include_extensions and isclass(model) and issubclass(model, Resource):
for extension_model in model.get_extension_models().values():
yield from iter_model_paths(extension_model)
