Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 43 additions & 0 deletions doc/tutorial.rst
Original file line number Diff line number Diff line change
Expand Up @@ -131,6 +131,49 @@ If a response resource type cannot be found, a `pydantic.ValidationError` will b
assert isinstance(group, Group)


Schema extensions
=================

:rfc:`7643 §3.3 <7643#section-3.3>` extensions are supported.
Extensions must be passed as resource type parameter, e.g. ``user = User[EnterpriseUser]`` or ``user = User[EnterpriseUser, SuperHero]``.
Extensions attributes are accessed with brackets, e.g. ``user[EnterpriseUser].employee_number``.

.. code-block:: python

import datetime
from pydantic_scim2 import User, EnterpriseUser, Meta

user = User[EnterpriseUser](
id="2819c223-7f76-453a-919d-413861904646",
user_name="bjensen@example.com",
meta=Meta(
resource_type="User",
created=datetime.datetime(
2010, 1, 23, 4, 56, 22, tzinfo=datetime.timezone.utc
),
),
)
user[EnterpriseUser].employee_number = "701984"
dump = user.model_dump(exclude_none=True, by_alias=True, mode="json")
assert dump == {
"schemas": [
"urn:ietf:params:scim:schemas:core:2.0:User",
"urn:ietf:params:scim:schemas:extension:enterprise:2.0:User"
],
"id": "2819c223-7f76-453a-919d-413861904646",
"meta": {
"resourceType": "User",
"created": "2010-01-23T04:56:22Z"
},
"userName": "bjensen@example.com",
"urn:ietf:params:scim:schemas:extension:enterprise:2.0:User": {
"schemas": [
"urn:ietf:params:scim:schemas:extension:enterprise:2.0:User"
],
"employeeNumber": "701984"
}
}

Custom models
=============

Expand Down
3 changes: 3 additions & 0 deletions pydantic_scim2/rfc7643/enterprise_user.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
from typing import List
from typing import Optional

from pydantic import AnyUrl
Expand All @@ -21,6 +22,8 @@ class Manager(SCIM2Model):


class EnterpriseUser(SCIM2Model):
schemas: List[str] = ["urn:ietf:params:scim:schemas:extension:enterprise:2.0:User"]

employee_number: Optional[str] = None
"""Numeric or alphanumeric identifier assigned to a person, typically based
on order of hire or association with anorganization."""
Expand Down
57 changes: 56 additions & 1 deletion pydantic_scim2/rfc7643/resource.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,19 @@
from datetime import datetime
from typing import Any
from typing import Generic
from typing import List
from typing import Optional
from typing import TypeVar

from pydantic import ConfigDict
from pydantic import field_serializer
from pydantic import model_validator
from typing_extensions import Self

from ..base import SCIM2Model

T = TypeVar("T", SCIM2Model, Any)


class Meta(SCIM2Model):
"""All "meta" sub-attributes are assigned by the service provider (have a
Expand Down Expand Up @@ -62,7 +70,7 @@ class Meta(SCIM2Model):
"""


class Resource(SCIM2Model):
class Resource(SCIM2Model, Generic[T]):
model_config = ConfigDict(extra="allow")

schemas: List[str]
Expand Down Expand Up @@ -124,3 +132,50 @@ class Resource(SCIM2Model):

meta: Meta
"""A complex attribute containing resource metadata."""

def __getitem__(self, item):
schema = item.model_fields["schemas"].default[0]

if not hasattr(self, schema):
setattr(self, schema, item())

return getattr(self, schema)

def __setitem__(self, item: type, value: "Resource"):
schema = item.model_fields["schemas"].default[0]
setattr(self, schema, value)

@model_validator(mode="after")
def load_model_extensions(self) -> Self:
"""Instanciate schema objects if found in the payload."""

extension_models = self.__pydantic_generic_metadata__.get("args")
if not extension_models:
return self

main_schema = self.model_fields["schemas"].default[0]
by_schema = {
ext.model_fields["schemas"].default[0]: ext for ext in extension_models
}
for schema in self.schemas:
if schema == main_schema:
continue

model = by_schema[schema]
if payload := getattr(self, schema, None):
setattr(self, schema, model.model_validate(payload))

return self

@field_serializer("schemas")
def set_extension_schemas(self, schemas: List[str]):
"""Add model extensions to 'schemas'."""

extension_models = self.__pydantic_generic_metadata__.get("args")
extension_schemas = [
ext.model_fields["schemas"].default[0] for ext in extension_models
]
schemas = self.schemas + [
schema for schema in extension_schemas if schema not in self.schemas
]
return schemas
28 changes: 12 additions & 16 deletions tests/test_enterprise_user.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

from pydantic_scim2 import Address
from pydantic_scim2 import Email
from pydantic_scim2 import EnterpriseUser
from pydantic_scim2 import Im
from pydantic_scim2 import PhoneNumber
from pydantic_scim2 import Photo
Expand All @@ -12,7 +13,7 @@

def test_enterprise_user(load_sample):
payload = load_sample("rfc7643-8.3-user-enterprise_user.json")
obj = User.model_validate(payload)
obj = User[EnterpriseUser].model_validate(payload)

assert obj.schemas == [
"urn:ietf:params:scim:schemas:core:2.0:User",
Expand Down Expand Up @@ -123,21 +124,16 @@ def test_enterprise_user(load_sample):
== "https://example.com/v2/Users/2819c223-7f76-453a-919d-413861904646"
)

# TODO: implement assertions for this
# "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User": {
# "employeeNumber": "701984",
# "costCenter": "4130",
# "organization": "Universal Studios",
# "division": "Theme Park",
# "department": "Tour Operations",
# "manager": {
# "value": "26118915-6090-4610-87e4-49d8ca9f808d",
# # TODO: relative URL are not supported by pydantic. Is this an error in the spec?
# #"$ref": "../Users/26118915-6090-4610-87e4-49d8ca9f808d",
# "$ref": "https://example.com/v2/Users/26118915-6090-4610-87e4-49d8ca9f808d",
# "displayName": "John Smith",
# },
# },
assert obj[EnterpriseUser].employee_number == "701984"
assert obj[EnterpriseUser].cost_center == "4130"
assert obj[EnterpriseUser].organization == "Universal Studios"
assert obj[EnterpriseUser].division == "Theme Park"
assert obj[EnterpriseUser].department == "Tour Operations"
assert obj[EnterpriseUser].manager.value == "26118915-6090-4610-87e4-49d8ca9f808d"
assert obj[EnterpriseUser].manager.ref == AnyUrl(
"https://example.com/v2/Users/26118915-6090-4610-87e4-49d8ca9f808d"
)
assert obj[EnterpriseUser].manager.display_name == "John Smith"

assert (
obj.model_dump(
Expand Down
160 changes: 160 additions & 0 deletions tests/test_resource_extension.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,160 @@
import datetime

from pydantic_scim2 import EnterpriseUser
from pydantic_scim2 import Manager
from pydantic_scim2 import Meta
from pydantic_scim2 import User


def test_extension_getitem():
"""Test that an extension can be accessed and update with __getitem__"""

user = User[EnterpriseUser](
id="2819c223-7f76-453a-919d-413861904646",
user_name="bjensen@example.com",
meta=Meta(
resource_type="User",
created=datetime.datetime(
2010, 1, 23, 4, 56, 22, tzinfo=datetime.timezone.utc
),
last_modified=datetime.datetime(
2011, 5, 13, 4, 42, 34, tzinfo=datetime.timezone.utc
),
version='W\\/"a330bc54f0671c9"',
location="https://example.com/v2/Users/2819c223-7f76-453a-919d-413861904646",
),
)
user[EnterpriseUser].employee_number = "701984"
user[EnterpriseUser].cost_center = "4130"
user[EnterpriseUser].organization = "Universal Studios"
user[EnterpriseUser].division = "Theme Park"
user[EnterpriseUser].department = "Tour Operations"
user[EnterpriseUser].manager = Manager(
value="26118915-6090-4610-87e4-49d8ca9f808d",
ref="https://example.com/v2/Users/26118915-6090-4610-87e4-49d8ca9f808d",
display_name="John Smith",
)

expected_payload = {
"id": "2819c223-7f76-453a-919d-413861904646",
"meta": {
"created": "2010-01-23T04:56:22Z",
"lastModified": "2011-05-13T04:42:34Z",
"location": "https://example.com/v2/Users/2819c223-7f76-453a-919d-413861904646",
"resourceType": "User",
"version": 'W\\/"a330bc54f0671c9"',
},
"schemas": [
"urn:ietf:params:scim:schemas:core:2.0:User",
"urn:ietf:params:scim:schemas:extension:enterprise:2.0:User",
],
"urn:ietf:params:scim:schemas:extension:enterprise:2.0:User": {
"employeeNumber": "701984",
"costCenter": "4130",
"organization": "Universal Studios",
"division": "Theme Park",
"department": "Tour Operations",
"manager": {
"value": "26118915-6090-4610-87e4-49d8ca9f808d",
"$ref": "https://example.com/v2/Users/26118915-6090-4610-87e4-49d8ca9f808d",
"displayName": "John Smith",
},
"schemas": [
"urn:ietf:params:scim:schemas:extension:enterprise:2.0:User",
],
},
"userName": "bjensen@example.com",
}
assert (
user.model_dump(exclude_none=True, by_alias=True, mode="json")
== expected_payload
)


def test_extension_setitem():
"""Test that an extension can be set with __setitem__"""

user = User[EnterpriseUser](
id="2819c223-7f76-453a-919d-413861904646",
user_name="bjensen@example.com",
meta=Meta(
resource_type="User",
created=datetime.datetime(
2010, 1, 23, 4, 56, 22, tzinfo=datetime.timezone.utc
),
last_modified=datetime.datetime(
2011, 5, 13, 4, 42, 34, tzinfo=datetime.timezone.utc
),
version='W\\/"a330bc54f0671c9"',
location="https://example.com/v2/Users/2819c223-7f76-453a-919d-413861904646",
),
)
user[EnterpriseUser] = EnterpriseUser(
employee_number="701984",
cost_center="4130",
organization="Universal Studios",
division="Theme Park",
department="Tour Operations",
manager=Manager(
value="26118915-6090-4610-87e4-49d8ca9f808d",
ref="https://example.com/v2/Users/26118915-6090-4610-87e4-49d8ca9f808d",
display_name="John Smith",
),
)

expected_payload = {
"id": "2819c223-7f76-453a-919d-413861904646",
"meta": {
"created": "2010-01-23T04:56:22Z",
"lastModified": "2011-05-13T04:42:34Z",
"location": "https://example.com/v2/Users/2819c223-7f76-453a-919d-413861904646",
"resourceType": "User",
"version": 'W\\/"a330bc54f0671c9"',
},
"schemas": [
"urn:ietf:params:scim:schemas:core:2.0:User",
"urn:ietf:params:scim:schemas:extension:enterprise:2.0:User",
],
"urn:ietf:params:scim:schemas:extension:enterprise:2.0:User": {
"employeeNumber": "701984",
"costCenter": "4130",
"organization": "Universal Studios",
"division": "Theme Park",
"department": "Tour Operations",
"manager": {
"value": "26118915-6090-4610-87e4-49d8ca9f808d",
"$ref": "https://example.com/v2/Users/26118915-6090-4610-87e4-49d8ca9f808d",
"displayName": "John Smith",
},
"schemas": [
"urn:ietf:params:scim:schemas:extension:enterprise:2.0:User",
],
},
"userName": "bjensen@example.com",
}
assert (
user.model_dump(exclude_none=True, by_alias=True, mode="json")
== expected_payload
)


def test_extension_no_payload():
"""An extension is defined but there is no matching payload."""

payload = {
"id": "2819c223-7f76-453a-919d-413861904646",
"meta": {
"created": "2010-01-23T04:56:22Z",
"lastModified": "2011-05-13T04:42:34Z",
"location": "https://example.com/v2/Users/2819c223-7f76-453a-919d-413861904646",
"resourceType": "User",
"version": 'W\\/"a330bc54f0671c9"',
},
"schemas": [
"urn:ietf:params:scim:schemas:core:2.0:User",
"urn:ietf:params:scim:schemas:extension:enterprise:2.0:User",
],
"userName": "bjensen@example.com",
}

User[EnterpriseUser].model_validate(payload)