|
| 1 | +from collections.abc import Iterator |
1 | 2 | from enum import Enum |
2 | 3 | from inspect import isclass |
3 | 4 | from typing import Annotated |
4 | 5 | from typing import Any |
5 | 6 | from typing import Generic |
6 | 7 | from typing import TypeVar |
| 8 | +from typing import cast |
7 | 9 |
|
8 | 10 | from pydantic import BaseModel as PydanticBaseModel |
9 | 11 | from pydantic import Field |
@@ -79,6 +81,137 @@ def _resolved_field(resource_class: type[BaseModel], attr_name: str) -> str | No |
79 | 81 | return _find_field_name(resource_class, attr_name) |
80 | 82 |
|
81 | 83 |
|
| 84 | +_ENVELOPE_FIELDS = frozenset({"schemas"}) |
| 85 | +"""Fields that carry the payload rather than the state it describes.""" |
| 86 | + |
| 87 | + |
| 88 | +def _attribute_name(model: type[BaseModel], field_name: str) -> str: |
| 89 | + """Return the SCIM spelling of a field, as a path segment.""" |
| 90 | + return model.model_fields[field_name].serialization_alias or field_name |
| 91 | + |
| 92 | + |
| 93 | +def _asserted_sub_attributes(entries: Any) -> set[str]: |
| 94 | + """Return the sub-attributes the entries of a wanted state name.""" |
| 95 | + asserted: set[str] = set() |
| 96 | + for entry in entries or []: |
| 97 | + if isinstance(entry, BaseModel): |
| 98 | + asserted |= entry.model_fields_set |
| 99 | + return asserted |
| 100 | + |
| 101 | + |
| 102 | +def _projection(entries: Any, asserted: set[str]) -> list[Any]: |
| 103 | + """Reduce the entries of a multi-valued attribute to what is worth comparing. |
| 104 | +
|
| 105 | + :rfc:`RFC7643 §2.4 <7643#section-2.4>` gives no significance to the order of |
| 106 | + a multi-valued attribute, so the projections are sorted before comparison. |
| 107 | + """ |
| 108 | + projected = [ |
| 109 | + tuple(sorted((name, getattr(entry, name, None)) for name in asserted)) |
| 110 | + if isinstance(entry, BaseModel) |
| 111 | + else entry |
| 112 | + for entry in entries or [] |
| 113 | + ] |
| 114 | + return sorted(projected, key=repr) |
| 115 | + |
| 116 | + |
| 117 | +def _operation( |
| 118 | + path: str, old: Any, new: Any, mutability: Mutability | None |
| 119 | +) -> tuple["PatchOperation.Op", str, Any]: |
| 120 | + """Return the operation writing *new* where the current state holds *old*. |
| 121 | +
|
| 122 | + Called once a difference is established. :rfc:`RFC7644 §3.5.2.3 |
| 123 | + <7644#section-3.5.2.3>` has a service provider treat a ``replace`` on an |
| 124 | + unset target as an ``add``, so a single operation covers both. An immutable |
| 125 | + attribute is the exception: :rfc:`RFC7644 §3.5.2 <7644#section-3.5.2>` lets |
| 126 | + a client add a value to one that had none, and nothing else. |
| 127 | + """ |
| 128 | + if mutability == Mutability.immutable: |
| 129 | + if old is not None: |
| 130 | + raise MutabilityException( |
| 131 | + attribute=path, mutability="immutable", operation="replace" |
| 132 | + ) |
| 133 | + return PatchOperation.Op.add, path, new |
| 134 | + |
| 135 | + if new is None or new == []: |
| 136 | + return PatchOperation.Op.remove, path, None |
| 137 | + |
| 138 | + return PatchOperation.Op.replace_, path, new |
| 139 | + |
| 140 | + |
| 141 | +def _diff_multi_valued( |
| 142 | + path: str, old: Any, new: Any, mutability: Mutability | None |
| 143 | +) -> Iterator[tuple["PatchOperation.Op", str, Any]]: |
| 144 | + """Diff a multi-valued attribute, which is replaced as a whole. |
| 145 | +
|
| 146 | + Only the sub-attributes the wanted entries name take part in the |
| 147 | + comparison, so the sub-attributes the peer alone maintains do not read as a |
| 148 | + difference. When the collection does change it is replaced entirely: |
| 149 | + :rfc:`RFC7643 §2.4 <7643#section-2.4>` gives the entries no identity, so an |
| 150 | + entry that changed cannot be told from a removed one and an added one. |
| 151 | + """ |
| 152 | + asserted = _asserted_sub_attributes(new) |
| 153 | + if _projection(old, asserted) == _projection(new, asserted): |
| 154 | + return |
| 155 | + |
| 156 | + yield _operation(path, old, new, mutability) |
| 157 | + |
| 158 | + |
| 159 | +def _diff_sub_object( |
| 160 | + prefix: str, |
| 161 | + path: str, |
| 162 | + old: Any, |
| 163 | + new: Any, |
| 164 | + mutability: Mutability | None, |
| 165 | +) -> Iterator[tuple["PatchOperation.Op", str, Any]]: |
| 166 | + """Diff a complex attribute or an extension, one sub-attribute at a time.""" |
| 167 | + if new is not None: |
| 168 | + yield from _diff(old, new, prefix) |
| 169 | + return |
| 170 | + |
| 171 | + if old is not None: |
| 172 | + yield _operation(path, old, None, mutability) |
| 173 | + |
| 174 | + |
| 175 | +def _diff( |
| 176 | + before: Any, after: Any, prefix: str = "" |
| 177 | +) -> Iterator[tuple["PatchOperation.Op", str, Any]]: |
| 178 | + """Yield the operations turning *before* into *after*. |
| 179 | +
|
| 180 | + Only the attributes *after* names are candidates: what a wanted state never |
| 181 | + mentions is left to the peer. Attributes are visited in declaration order, |
| 182 | + so a diff is reproducible. |
| 183 | + """ |
| 184 | + model = type(after) |
| 185 | + info = model.__scim_info__ |
| 186 | + for field_name in model.model_fields: |
| 187 | + if field_name not in after.model_fields_set: |
| 188 | + continue |
| 189 | + |
| 190 | + if field_name in _ENVELOPE_FIELDS: |
| 191 | + continue |
| 192 | + |
| 193 | + mutability = model.get_field_annotation(field_name, Mutability) |
| 194 | + if mutability == Mutability.read_only: |
| 195 | + continue |
| 196 | + |
| 197 | + old = getattr(before, field_name, None) if before is not None else None |
| 198 | + new = getattr(after, field_name, None) |
| 199 | + path = f"{prefix}{_attribute_name(model, field_name)}" |
| 200 | + |
| 201 | + if model.get_field_multiplicity(field_name): |
| 202 | + yield from _diff_multi_valued(path, old, new, mutability) |
| 203 | + |
| 204 | + elif field_name in info.extensions: |
| 205 | + urn = info.attribute_urns[field_name] |
| 206 | + yield from _diff_sub_object(f"{urn}:", urn, old, new, mutability) |
| 207 | + |
| 208 | + elif field_name in info.complex_fields: |
| 209 | + yield from _diff_sub_object(f"{path}.", path, old, new, mutability) |
| 210 | + |
| 211 | + elif old != new: |
| 212 | + yield _operation(path, old, new, mutability) |
| 213 | + |
| 214 | + |
82 | 215 | class PatchOperation(ComplexAttribute, Generic[ResourceT]): |
83 | 216 | class Op(str, Enum): |
84 | 217 | replace_ = "replace" |
@@ -362,6 +495,56 @@ def validate_operations(self, info: ValidationInfo) -> Self: |
362 | 495 |
|
363 | 496 | return self |
364 | 497 |
|
| 498 | + @classmethod |
| 499 | + def build_from( |
| 500 | + cls, before: ResourceT, after: ResourceT |
| 501 | + ) -> "PatchOp[ResourceT] | None": |
| 502 | + """Build the patch turning a resource state into another one. |
| 503 | +
|
| 504 | + Only the attributes *after* names take part in the comparison: what a |
| 505 | + wanted state never mentions is left to the peer, which is what |
| 506 | + distinguishes a patch from the :meth:`~scim2_models.Resource.replace` |
| 507 | + it stands for. An attribute named with no value is removed, as |
| 508 | + ``title=None`` reads as "clear the title" where an unnamed ``title`` |
| 509 | + reads as "leave it alone". |
| 510 | +
|
| 511 | + A multi-valued attribute is replaced as a whole, and only the |
| 512 | + sub-attributes the wanted entries name decide whether it changed. |
| 513 | + Read-only attributes never appear in the patch. |
| 514 | +
|
| 515 | + >>> from scim2_models import PatchOp, User |
| 516 | + >>> patch = PatchOp.build_from(User(nick_name="Barb"), User(nick_name="Babs")) |
| 517 | + >>> patch.model_dump()["Operations"] |
| 518 | + [{'op': 'replace', 'path': 'nickName', 'value': 'Babs'}] |
| 519 | +
|
| 520 | + :param before: The state the peer is believed to hold. |
| 521 | + :param after: The state the peer should hold. |
| 522 | + :return: The patch to send, or :data:`None` when the two states agree. |
| 523 | + :raises MutabilityException: If an immutable attribute already holding a |
| 524 | + value would be modified. |
| 525 | + :raises TypeError: If the two states are not of the same resource type. |
| 526 | + """ |
| 527 | + if type(before) is not type(after): |
| 528 | + raise TypeError( |
| 529 | + "Cannot compare two states of different types: " |
| 530 | + f"{type(before).__name__} and {type(after).__name__}" |
| 531 | + ) |
| 532 | + |
| 533 | + # Subscripted through the call the syntax stands for: mypy reads the |
| 534 | + # index of a generic as a type, not as a value. |
| 535 | + model = type(after) |
| 536 | + operation_class: Any = PatchOperation.__class_getitem__(model) |
| 537 | + path_class = Path.__class_getitem__(model) |
| 538 | + operations = [ |
| 539 | + operation_class(op=op, path=path_class(path), value=value) |
| 540 | + for op, path, value in _diff(before, after) |
| 541 | + ] |
| 542 | + if not operations: |
| 543 | + return None |
| 544 | + |
| 545 | + patch_class = PatchOp.__class_getitem__(model) |
| 546 | + return cast("PatchOp[ResourceT]", patch_class(operations=operations)) |
| 547 | + |
365 | 548 | def patch(self, resource: ResourceT, scim_policy: ScimPolicy | None = None) -> bool: |
366 | 549 | """Apply all PATCH operations to the given SCIM resource in sequence. |
367 | 550 |
|
|
0 commit comments