from collections.abc import Iterable
from collections.abc import Iterator
from collections.abc import Mapping
from dataclasses import dataclass
from dataclasses import field
from enum import Enum
from inspect import isclass
from typing import Any
from pydantic import BaseModel
from scim2_models import Attribute as SchemaAttribute
from scim2_models import AttributeBinding
from scim2_models import Mutability
from scim2_models import Path
from scim2_models import Resource
from scim2_models import ResponseParameters
from scim2_models import Schema
from scim2_models import Uniqueness
from sqlalchemy import ColumnElement
from sqlalchemy import Enum as SqlEnum
from sqlalchemy import String
from sqlalchemy import cast
from sqlalchemy import inspect
from sqlalchemy import literal
from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import InstrumentedAttribute
from sqlalchemy.orm import Mapper
from sqlalchemy.orm import RelationshipProperty
from sqlalchemy.orm import strategy_options
class _Storage(Enum):
"""How a mapping stores an attribute.
A derived attribute is read from another resource, such as the display of
a member, so writing it changes nothing. A ref is the $ref of a link,
built from its value: clients may send it, and writing it changes nothing
either. A link type is the type a Link gives its entries: clients may send
it, and the storage checks it against the linked resource.
"""
written = "written"
read_only = "read only"
derived = "derived"
ref = "ref"
link_type = "link type"
def _holds_text(expression: Any) -> bool:
try:
return expression.type.python_type is str
except NotImplementedError:
return True
[docs]
class Attribute:
"""A column holding a SCIM attribute that the storage only reads, or only writes.
To store a value in another form, map the attribute to a
:class:`~sqlalchemy.ext.hybrid.hybrid_property` that converts it, in Python
and in SQL.
:param expression: The column, or the
:class:`~sqlalchemy.ext.hybrid.hybrid_property`, holding the value.
:param readable: Whether the storage reads the attribute. When
:data:`False`, the attribute is write-only, such as a password: the
storage never returns it, nor filters or sorts on it.
:param writable: Whether the storage writes the attribute. When
:data:`False`, the attribute is read-only.
"""
def __init__(
self, expression: Any, *, readable: bool = True, writable: bool = True
) -> None:
self.expression = expression
self.readable = readable
self.writable = writable
[docs]
class Many:
"""The entries of a multi-valued complex attribute, stored in a child table.
The entries belong to the resource: writing the resource replaces them.
The relationship needs ``cascade="all, delete-orphan"``
(:ref:`sqlalchemy:cascade_delete_orphan`). Its ``order_by`` gives the
order of the entries. A sort uses the ``primary`` entry, or else
the first one.
:param relationship: The :func:`~sqlalchemy.orm.relationship` from the
resource record to the entry records.
:param attributes: The columns of the entry records, by sub-attribute name,
such as ``"value"`` or ``"type"``.
"""
def __init__(self, relationship: Any, attributes: Mapping[str, Any]) -> None:
self.relationship = relationship
self.attributes = attributes
[docs]
class Link:
"""A complex attribute linking to other resources.
A multi-valued attribute, such as the ``members`` of a group, maps to a
relationship holding several records. A single-valued one, such as the
``manager`` of an enterprise user, maps to a relationship holding one
record.
``value`` is the ``id`` of the linked resource. Its display name, the
``display`` or ``displayName`` sub-attribute, is the ``displayName`` of the
linked resource when its mapping has one, or else its ``userName``.
Writing the resource changes the links, and never creates nor deletes the
linked resources.
A Link links to the resources of one resource type. An attribute linking
to several resource types, such as the ``members`` of a group holding users
and groups, maps to a list of Links, one per resource type. The entries are
read in the order of the list.
:param relationship: The :func:`~sqlalchemy.orm.relationship` from the
resource record to the records of the linked resources.
:param resource_type: The name of the resource type of the linked
resources, as the storage knows it.
:param type: The ``type`` sub-attribute of the entries, such as ``"User"``
for the members of a group, or ``"direct"`` for the groups of a user.
A client may send it, and a write with another value is refused.
Without it, the storage does not store ``type``.
"""
def __init__(
self, relationship: Any, resource_type: str, *, type: str | None = None
) -> None:
self.relationship = relationship
self.resource_type = resource_type
self.type = type
@dataclass(eq=False)
class _Column:
"""A SCIM attribute stored in a column or a hybrid property.
textual tells whether filters compare the column as text, as they do for
an integer identifier.
"""
binding: AttributeBinding
expression: ColumnElement[Any]
key: str
readable: bool
writable: bool
textual: bool = False
@property
def is_string(self) -> bool:
target = self.binding.target_type
return (
target is not None
and SchemaAttribute.Type.from_python(target) == SchemaAttribute.Type.string
)
@property
def casefolded(self) -> bool:
"""Whether the values compare without their case."""
return self.is_string and not self.binding.case_exact
@property
def nullable(self) -> bool:
"""Whether the column can hold no value. A hybrid property is taken as nullable."""
columns = getattr(getattr(self.expression, "property", None), "columns", None)
return columns is None or bool(columns[0].nullable)
@property
def enumeration(self) -> SqlEnum | None:
"""The type of the column when it is an Enum, which stores strings."""
sql_type = self.expression.type
return sql_type if isinstance(sql_type, SqlEnum) else None
@property
def compared(self) -> ColumnElement[Any]:
"""The expression filters and sorts compare.
An Enum column is compared as text. PostgreSQL applies no string
function to its native enumerations, and sorts them in the order of
their declaration.
"""
if self.textual or self.enumeration is not None:
return cast(self.expression, String)
return self.expression
@staticmethod
def stored_string(enumeration: SqlEnum, member: Enum) -> str:
"""Return the string an Enum column stores for a member: its name, or its value with values_callable."""
strings: dict[Enum, str] = dict(
zip(type(member), enumeration.enums, strict=False)
)
return strings[member]
@dataclass(eq=False)
class _Collection:
"""A complex attribute stored in related records.
The entries of an owned collection belong to the resource. The others are
links to the resources of another mapping, the target. A single collection
holds one entry at most, such as the manager of a user.
link_type is the type a Link gives its entries, and type_column the
constant filters compare it as.
"""
binding: AttributeBinding
relationship: InstrumentedAttribute[Any]
columns: dict[str, _Column] = field(default_factory=dict)
resource_type: str | None = None
target: "ResourceMapping | None" = None
link_type: str | None = None
type_column: _Column | None = None
@property
def owned(self) -> bool:
return self.resource_type is None
@property
def entry_class(self) -> type[Any]:
return self.relationship_property.mapper.class_
@property
def relationship_property(self) -> RelationshipProperty[Any]:
return self.relationship.property # type: ignore[return-value]
@property
def single(self) -> bool:
return not self.binding.is_multivalued
def column(self, sub_field_name: str | None) -> _Column | None:
"""Return the column of a sub-attribute, or None for the collection named alone."""
if sub_field_name is None:
return None
if sub_field_name == "type" and self.type_column is not None:
return self.type_column
return self.columns.get(sub_field_name)
def exists(self, condition: Any = None) -> ColumnElement[bool]:
"""Return the condition that an entry exists, and matches condition when given."""
method = self.relationship.has if self.single else self.relationship.any
return method() if condition is None else method(condition)
@property
def writable(self) -> bool:
"""Whether writing the resource writes these entries."""
mutability = self.binding.model.get_field_annotation(
self.binding.field_name, Mutability
)
return bool(mutability != Mutability.read_only)
[docs]
class ResourceMapping:
"""How the resources of a SCIM model are stored in the records of a SQLAlchemy model.
The paths are checked against the SCIM model when the mapping is created,
so a mistake raises at import time rather than at the first request.
:param model: The SCIM model of the resources, such as
``User[EnterpriseUser]``.
:param record: The SQLAlchemy model of the records.
:param attributes: The attributes, by SCIM path. A value is a column, a
:class:`~sqlalchemy.ext.hybrid.hybrid_property`, an :class:`Attribute`, a :class:`Many`, a
:class:`Link`, or a list of :class:`Link`. ``id``, ``meta.created`` and ``meta.lastModified``
are required.
:param version: The ``version_id_col`` of the mapper of ``record``, which
gives ``meta.version`` (:ref:`sqlalchemy:mapper_version_counter`).
:raises ValueError: When a path is not valid for the model, or when a
value does not fit the attribute of its path.
"""
_REQUIRED = ("id", "meta.created", "meta.lastModified")
def __init__(
self,
model: type[Resource[Any]],
record: type[Any],
attributes: Mapping[str, Any],
*,
version: Any,
) -> None:
self.model = model
self.record = record
self._columns: dict[tuple[type[BaseModel], str, str | None], _Column] = {}
self._collections: dict[tuple[type[BaseModel], str], list[_Collection]] = {}
for required in self._REQUIRED:
if required not in attributes:
raise ValueError(f"The mapping of {model.__name__} needs {required!r}")
for path, value in attributes.items():
self._add(path, value)
mapper: Mapper[Any] = inspect(record)
columns = getattr(getattr(version, "property", None), "columns", [None])
if mapper.version_id_col is None or mapper.version_id_col is not columns[0]:
raise ValueError(
f"The version of {model.__name__} must be the version_id_col "
f"of the mapper of {record.__name__}"
)
self._version: InstrumentedAttribute[Any] = version
self._id = self._column("id")
self._id.textual = not _holds_text(self._id.expression)
self._created = self._column("meta.created")
self._last_modified = self._column("meta.lastModified")
[docs]
def schemas(self) -> list[Schema]:
"""Return the schemas of the resources, restricted to what the mapping stores.
They hold the mapped attributes only, and an attribute the mapping
reads but never writes is ``readOnly``. Publish them, rather than the
schemas of :attr:`model`, so that clients know what the server stores:
build the provider of the server with
:meth:`ScimProvider.from_discovery <scim2_models.ScimProvider.from_discovery>`,
and pass it to the storage.
:returns: The schema of the resource, then the schemas of its extensions.
"""
models: list[Any] = [self.model, *self.model.get_extension_models().values()]
return [self._restrict(model.to_schema()) for model in models]
def _restrict(self, schema: Schema) -> Schema:
attributes = []
for attribute in schema.attributes or []:
restricted = self._restrict_attribute(
f"{schema.id}:{attribute.name}", attribute
)
if restricted is not None:
attributes.append(restricted)
return schema.model_copy(update={"attributes": attributes})
def _restrict_attribute(self, urn: str, attribute: Any) -> Any:
"""Return an attribute of a schema as the mapping stores it, or None when it stores nothing of it."""
if attribute.sub_attributes:
sub_attributes = [
restricted
for sub_attribute in attribute.sub_attributes
if (
restricted := self._restrict_attribute(
f"{urn}.{sub_attribute.name}", sub_attribute
)
)
is not None
]
if not sub_attributes:
return None
written = any(
sub_attribute.mutability != Mutability.read_only
for sub_attribute in sub_attributes
)
return attribute.model_copy(
update={
"sub_attributes": sub_attributes,
"mutability": attribute.mutability
if written
else Mutability.read_only,
}
)
storage = self._storage(urn)
if storage is None:
return None
if storage in (_Storage.written, _Storage.link_type):
return attribute
if storage == _Storage.ref:
linked = self._linked_resource_types(urn)
return attribute.model_copy(update={"reference_types": linked})
return attribute.model_copy(update={"mutability": Mutability.read_only})
def _storage(self, urn: str) -> _Storage | None:
"""Return how the mapping stores the attribute of a URN, or None when it does not store it."""
key = urn.lower()
for column in self._columns.values():
if column.binding.urn.lower() == key:
return _Storage.written if column.writable else _Storage.read_only
for collection in self._all_collections():
if not collection.owned:
head = collection.binding.urn.lower()
if key == f"{head}.value":
return _Storage.written
if key in (f"{head}.display", f"{head}.displayname"):
return _Storage.derived
if key == f"{head}.$ref":
return _Storage.ref
if key == f"{head}.type" and collection.link_type is not None:
return _Storage.link_type
continue
for column in collection.columns.values():
if column.binding.urn.lower() == key:
return _Storage.written if column.writable else _Storage.read_only
return None
def _nullable(self, urn: str) -> bool:
"""Tell whether the column of an attribute can hold no value, True for an attribute without column."""
key = urn.lower()
return all(
column.nullable
for column in self._columns.values()
if column.binding.urn.lower() == key
)
def _linked_resource_types(self, urn: str) -> list[str]:
"""Return the resource types the Links of an attribute link to, from the URN of its $ref."""
head = urn.lower().removesuffix(".$ref")
return [
collection.resource_type
for collection in self._all_collections()
if collection.binding.urn.lower() == head
and collection.resource_type is not None
]
def _all_collections(self) -> Iterator[_Collection]:
"""Yield the collections of every attribute, each Link of an attribute being a collection."""
for collections in self._collections.values():
yield from collections
def _column(self, path: str) -> _Column:
binding = self._resolve(path)
return self._columns[
(binding.model, binding.field_name, binding.sub_field_name)
]
def _lookup(
self, binding: AttributeBinding
) -> list[tuple[_Collection | None, _Column | None]]:
"""Return each collection holding an attribute, if any, with the column of the attribute in it.
An attribute outside collections gives one pair, without collection.
An attribute linking to several resource types gives a pair for each
of its Links. The column is None for an attribute stored nowhere, and
for a collection named alone, such as emails.
"""
collections = self._collections.get((binding.model, binding.field_name))
if collections is None:
key = (binding.model, binding.field_name, binding.sub_field_name)
return [(None, self._columns.get(key))]
return [
(collection, collection.column(binding.sub_field_name))
for collection in collections
]
def _sub_columns(self, binding: AttributeBinding) -> list[_Column]:
"""Return the mapped sub-attributes of a complex attribute holding a single value."""
return [
column
for (model, field_name, sub_field_name), column in self._columns.items()
if model is binding.model
and field_name == binding.field_name
and sub_field_name is not None
]
def _string_columns(self) -> Iterator[_Column]:
"""Yield the columns holding strings, those of the collections included."""
for column in self._columns.values():
if column.is_string:
yield column
for collection in self._all_collections():
for column in collection.columns.values():
if column.is_string:
yield column
if collection.type_column is not None:
yield collection.type_column
def _unique_columns(self) -> Iterator[_Column]:
"""Yield the readable columns whose values no two resources share."""
for column in self._columns.values():
if column.readable and column.binding.get_annotation(Uniqueness) in (
Uniqueness.server,
Uniqueness.global_,
):
yield column
def _returned_collections(
self, parameters: ResponseParameters[Any] | None = None
) -> list[_Collection]:
"""Return the collections a response keeps, all of them without parameters."""
collections = list(self._all_collections())
if parameters is None:
return collections
paths = Path[self.model].iter_paths( # type: ignore[name-defined]
include_subattributes=False,
attributes=parameters.attributes,
excluded_attributes=parameters.excluded_attributes,
)
bindings = (path.resolve() for path in paths)
kept = {binding.urn for binding in bindings if binding is not None}
return [
collection for collection in collections if collection.binding.urn in kept
]
def _loader_options(
self, collections: Iterable[_Collection]
) -> list[strategy_options._AbstractLoad]:
"""Return the options loading collections along with the records.
Loading them up front is what lets an asynchronous session read them.
"""
return [
strategy_options.selectinload(collection.relationship)
for collection in collections
]
def _bind(self, mappings: Mapping[str, "ResourceMapping"]) -> None:
"""Bind each Link to the mapping of the resources it links to.
Several storages may share a mapping, as long as they link it to the
same mappings.
"""
for collection in self._all_collections():
if collection.owned:
continue
target = mappings.get(collection.resource_type) # type: ignore[arg-type]
if target is None:
raise ValueError(
f"No mapping for the resource type {collection.resource_type!r}"
)
if collection.target is not None and collection.target is not target:
raise ValueError(
f"{collection.binding.urn!r} already links to another mapping "
f"of {collection.resource_type!r}: a mapping serves storages "
"sharing the same mappings"
)
collection.target = target
collection.columns = {}
head = collection.binding.urn
for binding, column in (
(self._resolve(f"{head}.value"), target._id),
(
self._find(f"{head}.display") or self._find(f"{head}.displayName"),
target._display(),
),
):
if binding is None or column is None:
continue
collection.columns[binding.sub_field_name] = _Column( # type: ignore[index]
binding,
column.expression,
column.key,
column.readable,
False,
column.textual,
)
def _display(self) -> _Column | None:
"""Return the column a link to these resources displays."""
for path in ("displayName", "userName"):
binding = self._find(path)
if binding is None:
continue
((_, column),) = self._lookup(binding)
if column is not None and column.readable:
return column
return None
def _find(self, path: str) -> AttributeBinding | None:
return Path[self.model](path).resolve() # type: ignore[name-defined]
def _resolve(self, path: str) -> AttributeBinding:
binding = self._find(path)
if binding is None:
raise ValueError(f"{path!r} is not an attribute of {self.model.__name__}")
return binding
def _add(self, path: str, value: Any) -> None:
binding = self._resolve(path)
target = binding.target_type
is_complex = isclass(target) and issubclass(target, BaseModel)
if isinstance(value, Link):
value = [value]
if isinstance(value, Many) or (
isinstance(value, list) and all(isinstance(item, Link) for item in value)
):
if not (is_complex and binding.sub_field_name is None) or (
isinstance(value, Many) and not binding.is_multivalued
):
kind = (
"a multi-valued complex" if isinstance(value, Many) else "a complex"
)
raise ValueError(
f"{path!r} is not {kind} attribute of {self.model.__name__}"
)
if isinstance(value, Many):
collections = [self._many(path, binding, value)]
else:
collections = self._links(path, binding, value)
self._collections[(binding.model, binding.field_name)] = collections
return
if binding.is_multivalued or is_complex:
raise ValueError(
f"{path!r} holds several values or sub-attributes: map it with Many "
"or Link, or map its sub-attributes"
)
self._columns[(binding.model, binding.field_name, binding.sub_field_name)] = (
self._build_column(path, binding, value)
)
def _collection(
self, path: str, binding: AttributeBinding, relationship: Any
) -> _Collection:
"""Build the collection of a relationship, holding as many records as the attribute holds entries."""
collection = _Collection(binding, relationship)
if collection.relationship_property.uselist == collection.single:
raise ValueError(
f"{path!r} holds {'one entry' if collection.single else 'several entries'}, "
f"but its relationship holds {'several' if collection.single else 'one'}"
)
return collection
def _many(self, path: str, binding: AttributeBinding, many: Many) -> _Collection:
"""Build the collection of a Many, with the columns of its entries."""
collection = self._collection(path, binding, many.relationship)
if not collection.relationship_property.cascade.delete_orphan:
raise ValueError(
f"The relationship of {path!r} needs the delete-orphan cascade"
)
for sub_path, expression in many.attributes.items():
sub_binding = self._resolve(f"{path}.{sub_path}")
collection.columns[sub_binding.sub_field_name] = self._build_column( # type: ignore[index]
f"{path}.{sub_path}", sub_binding, expression
)
return collection
def _links(
self, path: str, binding: AttributeBinding, links: list[Link]
) -> list[_Collection]:
"""Build the collections of the Links of an attribute, one per resource type.
The type of the entries is compared, in filters, as a constant of each
Link.
"""
resource_types = [link.resource_type for link in links]
if not links or len(set(resource_types)) != len(resource_types):
raise ValueError(
f"{path!r} needs one Link per resource type, and at least one"
)
typed = {link.type is not None for link in links}
if len(typed) > 1:
raise ValueError(f"Either every Link of {path!r} has a type, or none")
type_binding = self._find(f"{path}.type")
if typed == {True} and type_binding is None:
raise ValueError(f"{path!r} has no type sub-attribute")
collections = []
for link in links:
collection = self._collection(path, binding, link.relationship)
collection.resource_type = link.resource_type
if link.type is not None:
collection.link_type = link.type
collection.type_column = _Column(
type_binding, # type: ignore[arg-type]
literal(link.type),
"type",
True,
False,
)
collections.append(collection)
return collections
def _build_column(
self, path: str, binding: AttributeBinding, value: Any
) -> _Column:
"""Build the column of an attribute, and decide whether it is written.
A read-only attribute, or a hybrid property without a setter, is never
written.
"""
attribute = value if isinstance(value, Attribute) else Attribute(value)
expression = attribute.expression
key = getattr(expression, "key", None)
if not isinstance(key, str):
raise ValueError(
f"{path!r} must be mapped to a column or a hybrid property"
)
writable = attribute.writable
descriptor = inspect(expression.class_).all_orm_descriptors[key]
mutabilities = {
binding.model.get_field_annotation(binding.field_name, Mutability),
binding.get_annotation(Mutability),
}
if Mutability.read_only in mutabilities or (
isinstance(descriptor, hybrid_property) and descriptor.fset is None
):
writable = False
return _Column(binding, expression, key, attribute.readable, writable)
