import warnings
from inspect import isclass
from typing import Any
from typing import Optional
from typing import get_args
from typing import get_origin
from pydantic import AliasGenerator
from pydantic import BaseModel as PydanticBaseModel
from pydantic import ConfigDict
from pydantic import FieldSerializationInfo
from pydantic import SerializationInfo
from pydantic import SerializerFunctionWrapHandler
from pydantic import ValidationInfo
from pydantic import ValidatorFunctionWrapHandler
from pydantic import field_serializer
from pydantic import field_validator
from pydantic import model_serializer
from pydantic import model_validator
from pydantic_core import PydanticCustomError
from typing_extensions import Self
from scim2_models.annotations import Mutability
from scim2_models.annotations import Required
from scim2_models.annotations import Returned
from scim2_models.context import Context
from scim2_models.exceptions import MutabilityException
from scim2_models.utils import UNION_TYPES
from scim2_models.utils import _find_field_name
from scim2_models.utils import _normalize_attribute_name
from scim2_models.utils import _to_camel
def _short_attr_path(urn: str) -> str:
"""Extract the short attribute path from a full URN.
For URNs like ``urn:...:User:userName``, returns ``userName``.
For URNs like ``urn:...:User:name.familyName``, returns ``name.familyName``.
For short names like ``userName``, returns ``userName`` as-is.
"""
if ":" in urn:
return urn.rsplit(":", 1)[1]
return urn
def _attr_matches(requested: str, current_urn: str) -> bool:
"""Check if a single requested attribute matches the current field URN.
Supports short names (``userName``), dotted paths (``name.familyName``),
and full extension URNs. Handles parent/child relationships.
"""
req_lower = requested.lower()
if ":" in requested:
current_lower = current_urn.lower()
return (
current_lower == req_lower
or req_lower.startswith(current_lower + ":")
or req_lower.startswith(current_lower + ".")
or current_lower.startswith(req_lower + ".")
or current_lower.startswith(req_lower + ":")
)
current_short = _short_attr_path(current_urn).lower()
return (
current_short == req_lower
or current_short.startswith(req_lower + ".")
or req_lower.startswith(current_short + ".")
)
def _exact_attr_match(attrs: list[str], current_urn: str) -> bool:
"""Check if current_urn exactly matches any entry in attrs (case-insensitive).
Used for ``excludedAttributes`` matching and :attr:`Returned.request` checking,
where parent/child relationship should not apply.
"""
current_short = _short_attr_path(current_urn).lower()
for attr in attrs:
attr_lower = attr.lower()
if ":" in attr:
if current_urn.lower() == attr_lower:
return True
else:
if current_short == attr_lower:
return True
return False
def _is_attribute_requested(requested_attrs: list[str], current_urn: str) -> bool:
"""Check if an attribute should be included based on the requested attributes.
Returns True if:
- The current attribute is explicitly requested
- A sub-attribute of the current attribute is requested
- The current attribute is a sub-attribute of a requested attribute
"""
return any(_attr_matches(req, current_urn) for req in requested_attrs)
[docs]
class BaseModel(PydanticBaseModel):
"""Base Model for everything."""
model_config = ConfigDict(
alias_generator=AliasGenerator(
validation_alias=_normalize_attribute_name,
serialization_alias=_to_camel,
),
validate_assignment=True,
populate_by_name=True,
use_attribute_docstrings=True,
extra="forbid",
)
[docs]
@classmethod
def get_field_annotation(cls, field_name: str, annotation_type: type) -> Any:
"""Return the annotation of type 'annotation_type' of the field 'field_name'.
This method extracts SCIM-specific annotations from a field's metadata,
such as :class:`~scim2_models.Mutability`, :class:`~scim2_models.Required`,
or :class:`~scim2_models.Returned` annotations.
:return: The annotation instance if found, otherwise the annotation type's default value
>>> from scim2_models.resources.user import User
>>> from scim2_models.annotations import Mutability, Required
Get the mutability annotation of the 'id' field:
>>> mutability = User.get_field_annotation("id", Mutability)
>>> mutability
<Mutability.read_only: 'readOnly'>
Get the required annotation of the 'user_name' field:
>>> required = User.get_field_annotation("user_name", Required)
>>> required
<Required.true: True>
If no annotation is found, returns the default value:
>>> missing = User.get_field_annotation("display_name", Required)
>>> missing
<Required.false: False>
"""
field_metadata = cls.model_fields[field_name].metadata
default_value = getattr(annotation_type, "_default", None)
def annotation_type_filter(item: Any) -> bool:
return isinstance(item, annotation_type)
field_annotation = next(
filter(annotation_type_filter, field_metadata), default_value
)
return field_annotation
[docs]
@classmethod
def get_field_root_type(cls, attribute_name: str) -> type | None:
"""Extract the root type from a model field.
This method unwraps complex type annotations to find the underlying
type, removing Optional and List wrappers to get to the actual type
of the field's content.
:return: The root type of the field, or None if not found
>>> from scim2_models.resources.user import User
>>> from scim2_models.resources.group import Group
Simple type:
>>> User.get_field_root_type("user_name")
<class 'str'>
``Optional`` type unwraps to the underlying type:
>>> User.get_field_root_type("display_name")
<class 'str'>
``List`` type unwraps to the element type:
>>> User.get_field_root_type("emails") # doctest: +ELLIPSIS
<class 'scim2_models.resources.user.Email'>
``Optional[List[T]]`` unwraps to ``T``:
>>> Group.get_field_root_type("members") # doctest: +ELLIPSIS
<class 'scim2_models.resources.group.GroupMember'>
"""
attribute_type = cls.model_fields[attribute_name].annotation
# extract 'x' from 'Optional[x]'
if get_origin(attribute_type) in UNION_TYPES:
attribute_type = get_args(attribute_type)[0]
# extract 'x' from 'List[x]'
origin = get_origin(attribute_type)
if origin and isclass(origin) and issubclass(origin, list):
attribute_type = get_args(attribute_type)[0]
return attribute_type
[docs]
@classmethod
def get_field_multiplicity(cls, attribute_name: str) -> bool:
"""Indicate whether a field holds multiple values.
This method determines if a field is defined as a list type,
which indicates it can contain multiple values. It handles
Optional wrappers correctly.
:return: True if the field holds multiple values (is a list), False otherwise
>>> from scim2_models.resources.user import User
>>> User.get_field_multiplicity("user_name")
False
>>> User.get_field_multiplicity("emails")
True
"""
attribute_type = cls.model_fields[attribute_name].annotation
# extract 'x' from 'Optional[x]'
if get_origin(attribute_type) in UNION_TYPES:
attribute_type = get_args(attribute_type)[0]
origin = get_origin(attribute_type)
return isinstance(origin, type) and issubclass(origin, list)
[docs]
@field_validator("*")
@classmethod
def check_request_attributes_mutability(
cls, value: Any, info: ValidationInfo
) -> Any:
"""Check and fix that the field mutability is expected according to the requests validation context, as defined in :rfc:`RFC7643 §7 <7643#section-7>`."""
if (
not info.context
or not info.field_name
or not info.context.get("scim")
or not Context.is_request(info.context["scim"])
):
return value
context = info.context.get("scim")
mutability = cls.get_field_annotation(info.field_name, Mutability)
exc = PydanticCustomError(
"mutability_error",
"Field '{field_name}' has mutability '{field_mutability}' but this in not valid in {context} context",
{
"field_name": info.field_name,
"field_mutability": mutability,
"context": context.name.lower().replace("_", " "),
},
)
if (
context in (Context.RESOURCE_QUERY_REQUEST, Context.SEARCH_REQUEST)
and mutability == Mutability.write_only
):
raise exc
if (
context
in (Context.RESOURCE_CREATION_REQUEST, Context.RESOURCE_REPLACEMENT_REQUEST)
and mutability == Mutability.read_only
):
return None
return value
[docs]
@model_validator(mode="wrap")
@classmethod
def normalize_attribute_names(
cls, value: Any, handler: ValidatorFunctionWrapHandler, info: ValidationInfo
) -> Self:
"""Normalize payload attribute names.
:rfc:`RFC7643 §2.1 <7643#section-2.1>` indicate that attribute
names should be case-insensitive. Any attribute name is
transformed in lowercase so any case is handled the same way.
"""
def normalize_dict_keys(
input_dict: dict[str, Any], model_class: type["BaseModel"]
) -> dict[str, Any]:
"""Normalize dictionary keys, preserving case for Any fields."""
result = {}
for key, val in input_dict.items():
field_name = _find_field_name(model_class, key)
field_type = (
model_class.get_field_root_type(field_name) if field_name else None
)
# Don't normalize keys for attributes typed with Any
# This way, agnostic dicts such as PatchOp.operations.value
# are preserved
if field_name and field_type == Any:
result[key] = normalize_value(val)
else:
result[_normalize_attribute_name(key)] = normalize_value(
val, field_type
)
return result
def normalize_value(
val: Any, model_class: type["BaseModel"] | None = None
) -> Any:
"""Normalize input value based on model class."""
if not isinstance(val, dict):
return val
# If no model_class, preserve original keys
if not model_class:
return {k: normalize_value(v) for k, v in val.items()}
return normalize_dict_keys(val, model_class)
normalized_value = normalize_value(value, cls)
obj = handler(normalized_value)
assert isinstance(obj, cls)
return obj
[docs]
@model_validator(mode="wrap")
@classmethod
def check_response_attributes_returnability(
cls, value: Any, handler: ValidatorFunctionWrapHandler, info: ValidationInfo
) -> Self:
"""Check that the fields returnability is expected according to the responses validation context, as defined in :rfc:`RFC7643 §7 <7643#section-7>`."""
obj = handler(value)
assert isinstance(obj, cls)
if (
not info.context
or not info.context.get("scim")
or not Context.is_response(info.context["scim"])
):
return obj
for field_name in cls.model_fields:
returnability = cls.get_field_annotation(field_name, Returned)
if returnability == Returned.always and getattr(obj, field_name) is None:
raise PydanticCustomError(
"returned_error",
"Field '{field_name}' has returnability 'always' but value is missing or null",
{
"field_name": field_name,
},
)
if returnability == Returned.never and getattr(obj, field_name) is not None:
raise PydanticCustomError(
"returned_error",
"Field '{field_name}' has returnability 'never' but value is set",
{
"field_name": field_name,
},
)
return obj
[docs]
@model_validator(mode="wrap")
@classmethod
def check_response_attributes_necessity(
cls, value: Any, handler: ValidatorFunctionWrapHandler, info: ValidationInfo
) -> Self:
"""Check that the required attributes are present in creations and replacement requests."""
obj = handler(value)
assert isinstance(obj, cls)
if (
not info.context
or not info.context.get("scim")
or info.context["scim"]
not in (
Context.RESOURCE_CREATION_REQUEST,
Context.RESOURCE_REPLACEMENT_REQUEST,
)
):
return obj
for field_name in cls.model_fields:
necessity = cls.get_field_annotation(field_name, Required)
if necessity == Required.true and getattr(obj, field_name) is None:
raise PydanticCustomError(
"required_error",
"Field '{field_name}' is required but value is missing or null",
{
"field_name": field_name,
},
)
return obj
[docs]
@model_validator(mode="wrap")
@classmethod
def check_replacement_request_mutability(
cls, value: Any, handler: ValidatorFunctionWrapHandler, info: ValidationInfo
) -> Self:
"""Check if 'immutable' attributes have been mutated in replacement requests."""
from scim2_models.resources.resource import Resource
obj = handler(value)
assert isinstance(obj, cls)
context = info.context.get("scim") if info.context else None
original = info.context.get("original") if info.context else None
if (
context == Context.RESOURCE_REPLACEMENT_REQUEST
and issubclass(cls, Resource)
and original is not None
):
try:
obj._apply_replace_constraints(original)
except MutabilityException as exc:
raise exc.as_pydantic_error() from exc
return obj
[docs]
@model_validator(mode="after")
def check_primary_attribute_uniqueness(self, info: ValidationInfo) -> Self:
"""Validate that only one attribute can be marked as primary in multi-valued lists.
Per RFC 7643 Section 2.4: The primary attribute value 'true' MUST appear no more than once.
"""
scim_context = info.context.get("scim") if info.context else None
if not scim_context or scim_context == Context.DEFAULT:
return self
for field_name in self.__class__.model_fields:
if not self.get_field_multiplicity(field_name):
continue
field_value = getattr(self, field_name)
if field_value is None:
continue
element_type = self.get_field_root_type(field_name)
if (
element_type is None
or not isclass(element_type)
or not issubclass(element_type, PydanticBaseModel)
or "primary" not in element_type.model_fields
):
continue
primary_count = sum(
1 for item in field_value if getattr(item, "primary", None) is True
)
if primary_count > 1:
raise PydanticCustomError(
"primary_uniqueness_error",
"Field '{field_name}' has {count} items marked as primary, but only one is allowed per RFC 7643",
{
"field_name": field_name,
"count": primary_count,
},
)
return self
def _apply_replace_constraints(self, original: Self) -> None:
"""Enforce RFC 7644 §3.5.1 replace (PUT) semantics.
- ``readOnly`` fields are copied from *original* unconditionally.
- ``immutable`` fields are copied from *original* when absent from
``self``; a :class:`~scim2_models.MutabilityException` is raised
when the value differs.
Recursively applies to nested single-valued complex attributes.
"""
from .attributes import is_complex_attribute
for field_name in type(self).model_fields:
mutability = type(self).get_field_annotation(field_name, Mutability)
original_val = getattr(original, field_name)
if mutability == Mutability.read_only:
# RFC 7644 §3.5.1: "readOnly" values provided SHALL be ignored.
setattr(self, field_name, original_val)
elif mutability == Mutability.immutable:
self_val = getattr(self, field_name)
if self_val is None and original_val is not None:
# RFC 7643 §7: "SHALL NOT be updated" — omitting an
# immutable field is not a request to clear it.
setattr(self, field_name, original_val)
elif self_val != original_val:
# RFC 7644 §3.5.1: input values MUST match.
raise MutabilityException(
attribute=field_name, mutability="immutable"
)
attr_type = type(self).get_field_root_type(field_name)
if (
attr_type
and is_complex_attribute(attr_type)
and not type(self).get_field_multiplicity(field_name)
):
original_sub = getattr(original, field_name)
replacement_sub = getattr(self, field_name)
if original_sub is not None and replacement_sub is not None:
replacement_sub._apply_replace_constraints(original_sub)
def _set_complex_attribute_urns(self) -> None:
"""Navigate through attributes and sub-attributes of type ComplexAttribute, and mark them with a '_attribute_urn' attribute.
'_attribute_urn' will later be used by 'get_attribute_urn'.
"""
from .attributes import ComplexAttribute
from .attributes import is_complex_attribute
if isinstance(self, ComplexAttribute):
main_schema = self._attribute_urn
separator = "."
else:
main_schema = getattr(self.__class__, "__schema__", None)
if main_schema is None:
return
separator = ":"
for field_name in self.__class__.model_fields:
attr_type = self.get_field_root_type(field_name)
if not attr_type or not is_complex_attribute(attr_type):
continue
alias = (
self.__class__.model_fields[field_name].serialization_alias
or field_name
)
schema = f"{main_schema}{separator}{alias}"
if attr_value := getattr(self, field_name):
if isinstance(attr_value, list):
for item in attr_value:
item._attribute_urn = schema
else:
attr_value._attribute_urn = schema
[docs]
@field_serializer("*", mode="wrap")
def scim_serializer(
self,
value: Any,
handler: SerializerFunctionWrapHandler,
info: FieldSerializationInfo,
) -> Any:
"""Serialize the fields according to mutability indications passed in the serialization context."""
value = handler(value)
scim_ctx = info.context.get("scim") if info.context else None
if scim_ctx and Context.is_request(scim_ctx):
value = self._scim_request_serializer(value, info)
if scim_ctx and Context.is_response(scim_ctx):
value = self._scim_response_serializer(value, info)
return value
def _scim_request_serializer(self, value: Any, info: FieldSerializationInfo) -> Any:
"""Serialize the fields according to mutability indications passed in the serialization context."""
mutability = self.get_field_annotation(info.field_name, Mutability)
scim_ctx = info.context.get("scim") if info.context else None
if (
scim_ctx
in (Context.RESOURCE_CREATION_REQUEST, Context.RESOURCE_REPLACEMENT_REQUEST)
and mutability == Mutability.read_only
):
return None
if (
scim_ctx
in (
Context.RESOURCE_QUERY_REQUEST,
Context.SEARCH_REQUEST,
)
and mutability == Mutability.write_only
):
return None
return value
def _scim_response_serializer(
self, value: Any, info: FieldSerializationInfo
) -> Any:
"""Serialize the fields according to returnability indications passed in the serialization context."""
returnability = self.get_field_annotation(info.field_name, Returned)
attribute_urn = self.get_attribute_urn(info.field_name)
included_attrs = info.context.get("scim_attributes", []) if info.context else []
excluded_attrs = (
info.context.get("scim_excluded_attributes", []) if info.context else []
)
if returnability == Returned.never:
return None
if returnability == Returned.default and (
(
included_attrs
and not _is_attribute_requested(included_attrs, attribute_urn)
)
or _exact_attr_match(excluded_attrs, attribute_urn)
):
return None
if returnability == Returned.request and not _exact_attr_match(
included_attrs, attribute_urn
):
return None
return value
[docs]
@model_serializer(mode="wrap")
def model_serializer_exclude_none(
self, handler: SerializerFunctionWrapHandler, info: SerializationInfo
) -> dict[str, Any]:
"""Remove `None` values inserted by the :meth:`~scim2_models.base.BaseModel.scim_serializer`."""
self._set_complex_attribute_urns()
result = handler(self)
return {key: value for key, value in result.items() if value is not None}
[docs]
@classmethod
def model_validate(
cls,
*args: Any,
scim_ctx: Context | None = Context.DEFAULT,
original: Optional["BaseModel"] = None,
**kwargs: Any,
) -> Self:
"""Validate SCIM payloads and generate model representation by using Pydantic :meth:`~pydantic.BaseModel.model_validate`.
:param scim_ctx: The SCIM :class:`~scim2_models.Context` in which the validation happens.
:param original: If this parameter is set during :attr:`~Context.RESOURCE_REPLACEMENT_REQUEST`,
:attr:`~scim2_models.Mutability.immutable` parameters will be compared against the *original* model value.
An exception is raised if values are different.
.. deprecated:: 0.6.7
Use :meth:`replace` on the validated instance instead.
Will be removed in 0.8.0.
"""
if original is not None:
warnings.warn(
"The 'original' parameter is deprecated, "
"use the 'replace' method on the validated instance instead. "
"Will be removed in 0.8.0.",
DeprecationWarning,
stacklevel=2,
)
context = kwargs.setdefault("context", {})
context.setdefault("scim", scim_ctx)
context.setdefault("original", original)
return super().model_validate(*args, **kwargs)
[docs]
def get_attribute_urn(self, field_name: str) -> str:
"""Build the full URN of the attribute.
See :rfc:`RFC7644 §3.10 <7644#section-3.10>`.
"""
from scim2_models.resources.resource import Extension
main_schema = getattr(self.__class__, "__schema__", None)
field = self.__class__.model_fields[field_name]
alias = field.serialization_alias or field_name
field_type = self.get_field_root_type(field_name)
if isclass(field_type) and issubclass(field_type, Extension):
return alias
if main_schema is None:
return alias
return f"{main_schema}:{alias}"
