import copyreg
from datetime import datetime
from typing import TYPE_CHECKING
from typing import Annotated
from typing import Any
from typing import Generic
from typing import TypeVar
from typing import Union
from typing import get_args
from typing import get_origin
from pydantic import Field
from pydantic import SerializationInfo
from pydantic import SerializerFunctionWrapHandler
from pydantic import ValidationInfo
from pydantic import ValidatorFunctionWrapHandler
from pydantic import WrapSerializer
from pydantic import field_serializer
from pydantic import model_validator
from pydantic_core import PydanticCustomError
from typing_extensions import Self
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 ..attributes import is_complex_attribute
from ..base import BaseModel
from ..context import Context
from ..exceptions import InvalidPathException
from ..path import Path
from ..scim_object import ScimObject
from ..utils import UNION_TYPES
from ..utils import _normalize_attribute_name
if TYPE_CHECKING:
from .schema import Attribute
from .schema import Schema
[docs]
class Extension(ScimObject):
[docs]
@classmethod
def to_schema(cls) -> "Schema":
"""Build a :class:`~scim2_models.Schema` from the current extension class."""
return _model_to_schema(cls)
[docs]
@classmethod
def from_schema(cls, schema: "Schema") -> type["Extension"]:
"""Build a :class:`~scim2_models.Extension` subclass from the schema definition."""
from .schema import _make_python_model
return _make_python_model(schema, cls)
AnyExtension = TypeVar("AnyExtension", bound="Extension")
_PARAMETERIZED_CLASSES: dict[tuple[type, tuple[Any, ...]], type] = {}
def _extension_serializer(
value: Any, handler: SerializerFunctionWrapHandler, info: SerializationInfo
) -> dict[str, Any] | None:
"""Exclude the Resource attributes from the extension dump.
For instance, attributes 'meta', 'id' or 'schemas' should not be
dumped when the model is used as an extension for another model.
"""
if value is None:
return None
partial_result = handler(value)
result = {
attr_name: value
for attr_name, value in partial_result.items()
if attr_name not in Resource.model_fields
}
return result or None
[docs]
class Resource(ScimObject, Generic[AnyExtension]):
# Common attributes as defined by
# https://www.rfc-editor.org/rfc/rfc7643#section-3.1
id: Annotated[
str | None, Mutability.read_only, Returned.always, Uniqueness.global_
] = None
"""A unique identifier for a SCIM resource as defined by the service
provider.
id is mandatory is the resource representation, but is forbidden in
resource creation or replacement requests.
"""
external_id: Annotated[
str | None, Mutability.read_write, Returned.default, CaseExact.true
] = None
"""A String that is an identifier for the resource as defined by the
provisioning client."""
meta: Annotated[Meta | None, Mutability.read_only, Returned.default] = None
"""A complex attribute containing resource metadata."""
[docs]
def replace(self, original: Self) -> None:
"""Apply :rfc:`RFC 7644 ยง3.5.1 <7644#section-3.5.1>` replace (PUT) semantics.
``readOnly`` fields are copied from *original*.
``immutable`` fields are preserved from *original* when absent,
or checked for equality when present.
:param original: The original resource state to compare against.
:raises MutabilityException: If an immutable field value differs.
"""
self._apply_replace_constraints(original)
@classmethod
def __class_getitem__(cls, item: Any) -> type["Resource[Any]"]:
"""Create a Resource class with extension fields dynamically added."""
if hasattr(cls, "__scim_extension_metadata__"):
return cls
extensions = get_args(item) if get_origin(item) in UNION_TYPES else [item]
# Skip TypeVar parameters and Any (used for generic class definitions)
valid_extensions = [
extension
for extension in extensions
if not isinstance(extension, TypeVar) and extension is not Any
]
if not valid_extensions:
return cls
cache_key = (cls, tuple(valid_extensions))
if cache_key in _PARAMETERIZED_CLASSES:
return _PARAMETERIZED_CLASSES[cache_key]
for extension in valid_extensions:
if not (isinstance(extension, type) and issubclass(extension, Extension)):
raise TypeError(f"{extension} is not a valid Extension type")
class_name = (
f"{cls.__name__}[{', '.join(ext.__name__ for ext in valid_extensions)}]"
)
class_attrs = {"__scim_extension_metadata__": valid_extensions}
for extension in valid_extensions:
schema = extension.__schema__
class_attrs[extension.__name__] = Field(
default=None, # type: ignore[arg-type]
serialization_alias=schema,
validation_alias=_normalize_attribute_name(schema),
)
new_annotations = {
extension.__name__: Annotated[
extension | None,
WrapSerializer(_extension_serializer),
]
for extension in valid_extensions
}
new_class = type(
class_name,
(cls,),
{
"__annotations__": new_annotations,
**class_attrs,
},
)
_PARAMETERIZED_CLASSES[cache_key] = new_class
return new_class
def __getitem__(self, item: Any) -> Any:
"""Get a value by extension type or path.
:param item: An Extension subclass or a path (string or Path).
:returns: The extension instance or the value at the path.
:raises KeyError: If the path references a non-existent field.
Examples::
user[EnterpriseUser] # Get extension
user["userName"] # Get attribute
user["name.familyName"] # Get nested attribute
"""
if isinstance(item, type) and issubclass(item, Extension):
item = item.__schema__
bound_path = Path.__class_getitem__(type(self))
path = item if isinstance(item, Path) else bound_path(str(item))
try:
return path.get(self)
except InvalidPathException as exc:
raise KeyError(str(item)) from exc
def __setitem__(self, item: Any, value: Any) -> None:
"""Set a value by extension type or path.
:param item: An Extension subclass or a path (string or Path).
:param value: The value to set.
:raises KeyError: If the path references a non-existent field.
Examples::
user[EnterpriseUser] = EnterpriseUser(employee_number="123")
user["displayName"] = "John Doe"
user["name.familyName"] = "Doe"
"""
if isinstance(item, type) and issubclass(item, Extension):
item = item.__schema__
bound_path = Path.__class_getitem__(type(self))
path = item if isinstance(item, Path) else bound_path(str(item))
try:
path.set(self, value)
except InvalidPathException as exc:
raise KeyError(str(item)) from exc
def __delitem__(self, item: Any) -> None:
"""Delete a value by extension type or path.
:param item: An Extension subclass or a path (string or Path).
:raises KeyError: If the path references a non-existent field.
Examples::
del user[EnterpriseUser] # Remove extension
del user["displayName"] # Remove attribute
"""
if isinstance(item, type) and issubclass(item, Extension):
item = item.__schema__
bound_path = Path.__class_getitem__(type(self))
path = item if isinstance(item, Path) else bound_path(str(item))
try:
path.delete(self)
except InvalidPathException as exc:
raise KeyError(str(item)) from exc
[docs]
@classmethod
def get_extension_models(cls) -> dict[str, type[Extension]]:
"""Return extension a dict associating extension models with their schemas."""
extension_models = getattr(cls, "__scim_extension_metadata__", [])
by_schema: dict[str, type[Extension]] = {
ext.__schema__: ext for ext in extension_models
}
return by_schema
[docs]
@classmethod
def get_extension_model(
cls, name_or_schema: Union[str, "Schema"]
) -> type[Extension] | None:
"""Return an extension by its name or schema."""
for schema, extension in cls.get_extension_models().items():
if schema == name_or_schema or extension.__name__ == name_or_schema:
return extension
return None
[docs]
@staticmethod
def get_by_schema(
resource_types: list[type["Resource[Any]"]],
schema: str,
with_extensions: bool = True,
) -> type["Resource[Any]"] | type["Extension"] | None:
"""Given a resource type list and a schema, find the matching resource type."""
by_schema: dict[str, type[Resource[Any]] | type[Extension]] = {
getattr(resource_type, "__schema__", "").lower(): resource_type
for resource_type in (resource_types or [])
}
if with_extensions:
for resource_type in resource_types:
by_schema.update(
{
schema.lower(): extension
for schema, extension in resource_type.get_extension_models().items()
}
)
return by_schema.get(schema.lower())
[docs]
@staticmethod
def get_by_payload(
resource_types: list[type["Resource[Any]"]],
payload: dict[str, Any],
**kwargs: Any,
) -> type | None:
"""Given a resource type list and a payload, find the matching resource type."""
if not payload or not payload.get("schemas"):
return None
schema = payload["schemas"][0]
return Resource.get_by_schema(resource_types, schema, **kwargs)
[docs]
@field_serializer("schemas")
def set_extension_schemas(
self, schemas: Annotated[list[str], Required.true]
) -> list[str]:
"""Add model extension ids to the 'schemas' attribute."""
extension_schemas = self.get_extension_models().keys()
schemas = self.schemas + [
schema for schema in extension_schemas if schema not in self.schemas
]
return schemas
@model_validator(mode="wrap")
@classmethod
def _validate_extension_schemas(
cls, value: Any, handler: ValidatorFunctionWrapHandler, info: ValidationInfo
) -> Self:
"""Validate that extension schemas are known."""
obj: Self = handler(value)
scim_ctx = info.context.get("scim") if info.context else None
if scim_ctx is None or scim_ctx == Context.DEFAULT:
return obj
base_schema = getattr(cls, "__schema__", None)
if not base_schema:
return obj
allowed_extensions = set(cls.get_extension_models().keys())
provided_schemas = set(obj.schemas) - {base_schema}
unknown = provided_schemas - allowed_extensions
if unknown:
raise PydanticCustomError(
"unknown_extension_schema",
"Unknown extension schemas: {schemas}",
{"schemas": ", ".join(sorted(unknown))},
)
return obj
[docs]
@classmethod
def to_schema(cls) -> "Schema":
"""Build a :class:`~scim2_models.Schema` from the current resource class."""
return _model_to_schema(cls)
[docs]
@classmethod
def from_schema(cls, schema: "Schema") -> type["Resource[Any]"]:
"""Build a :class:`scim2_models.Resource` subclass from the schema definition."""
from .schema import _make_python_model
return _make_python_model(schema, cls)
AnyResource = TypeVar("AnyResource", bound="Resource[Any]")
def _dedicated_attributes(
model: type[BaseModel], excluded_models: list[type[BaseModel]]
) -> dict[str, Any]:
"""Return attributes that are not members the parent 'excluded_models'."""
def compare_field_infos(fi1: Any, fi2: Any) -> bool:
if not fi1 or not fi2:
return False
slot_names = copyreg._slotnames(type(fi1)) # type: ignore[attr-defined]
return all(getattr(fi1, attr) == getattr(fi2, attr) for attr in slot_names)
parent_field_infos = {
field_name: field_info
for excluded_model in excluded_models
for field_name, field_info in excluded_model.model_fields.items()
}
field_infos = {
field_name: field_info
for field_name, field_info in model.model_fields.items()
if not compare_field_infos(field_info, parent_field_infos.get(field_name))
}
return field_infos
def _model_to_schema(model: type[BaseModel]) -> "Schema":
from scim2_models.resources.schema import Schema
schema_urn = getattr(model, "__schema__", "") or ""
field_infos = _dedicated_attributes(model, [Resource])
attributes = [
_model_attribute_to_scim_attribute(model, attribute_name)
for attribute_name in field_infos
if attribute_name != "schemas"
]
schema = Schema(
name=model.__name__,
id=schema_urn,
description=model.__doc__ or model.__name__,
attributes=attributes,
)
return schema
def _model_attribute_to_scim_attribute(
model: type[BaseModel], attribute_name: str
) -> "Attribute":
from scim2_models.resources.schema import Attribute
field_info = model.model_fields[attribute_name]
root_type = model.get_field_root_type(attribute_name)
if root_type is None:
raise ValueError(
f"Could not determine root type for attribute {attribute_name}"
)
attribute_type = Attribute.Type.from_python(root_type)
sub_attributes = (
[
_model_attribute_to_scim_attribute(root_type, sub_attribute_name)
for sub_attribute_name in root_type.model_fields # type: ignore
if (
attribute_name != "sub_attributes"
or sub_attribute_name != "sub_attributes"
)
]
if root_type and is_complex_attribute(root_type)
else None
)
kwargs: dict[str, Any] = {
"name": field_info.serialization_alias or attribute_name,
"type": Attribute.Type(attribute_type),
"multi_valued": model.get_field_multiplicity(attribute_name),
"description": field_info.description,
"canonical_values": field_info.examples,
"required": model.get_field_annotation(attribute_name, Required),
"case_exact": model.get_field_annotation(attribute_name, CaseExact),
"mutability": model.get_field_annotation(attribute_name, Mutability),
"returned": model.get_field_annotation(attribute_name, Returned),
"sub_attributes": sub_attributes,
}
if attribute_type != Attribute.Type.complex:
kwargs["uniqueness"] = model.get_field_annotation(attribute_name, Uniqueness)
if attribute_type == Attribute.Type.reference:
kwargs["reference_types"] = root_type.get_scim_reference_types() # type: ignore[attr-defined]
return Attribute(**kwargs)
