Integrations

This section shows how to integrate scim2-models with your web framework to build a SCIM server.

Storage layer

For the sake of simplicity, all integration example will use the following simplistic storage layer. It wraps an in-memory dictionary and enforces business constraints such as userName uniqueness. In real applications, you will replace these functions with ORM calls (Django ORM, SQLAlchemy etc.), and adapt the code accordingly.

Minimalist storage layer

records = {}

MAX_RESULTS = 50


def get_record(record_id):
    """Return the record for *record_id*, raising KeyError if absent."""
    if record_id not in records:
        raise KeyError(record_id)
    return records[record_id]


def list_records():
    """Return every stored record."""
    return list(records.values())


# -- sorting-start --
def page_of(resources, req):
    """Return the total count and the page a query asks for.

    Sorting comes before paging, so a page holds the same resources whatever
    the order asked for. A page never exceeds ``MAX_RESULTS`` entries, which is
    the bound the :class:`~scim2_models.ServiceProviderConfig` advertises.

    :param resources: The SCIM resources to answer from.
    :param req: The parsed query.
    :return: A ``(total, page)`` tuple.
    """
    if req.sort_by:
        resources = sort_resources(resources, req.sort_by, req.sort_order)

    start = req.start_index_0 or 0
    limit = start + MAX_RESULTS
    stop = req.stop_index_0
    stop = limit if stop is None else min(stop, limit)
    return len(resources), resources[start:stop]


def sort_resources(resources, sort_by, sort_order=None):
    """Order resources by an attribute, per :rfc:`RFC7644 §3.4.2.3 <7644#section-3.4.2.3>`.

    :param resources: The SCIM resources to order.
    :param sort_by: The ``sortBy`` query parameter, resolved by the request it
        came from, which names the resource type the endpoint serves.
    :param sort_order: The ``sortOrder`` query parameter, ascending by default.
    :raises InvalidPathException: If the attribute is unknown.
    """
    if sort_by.field_name is None:
        raise InvalidPathException(
            path=str(sort_by), detail=f"Cannot sort on {sort_by!r}"
        )

    # "String type attributes are case insensitive by default, unless the
    # attribute type is defined as a case-exact string."
    case_exact = sort_by.model.get_field_annotation(sort_by.field_name, CaseExact)
    descending = sort_order == SearchRequest.SortOrder.descending

    def key(resource):
        value = sort_by.get(resource, strict=False)
        if isinstance(value, list):
            # "resources are sorted by the value of the primary attribute, if
            # any, or else the first value in the list, if any."
            primary = next((each for each in value if each.primary), None)
            entry = primary or (value[0] if value else None)
            value = entry.value if entry else None
        if isinstance(value, str) and not case_exact:
            value = value.casefold()
        # "if there is no data for the specified sortBy value, they are sorted
        # via the sortOrder parameter, i.e., they are ordered last if ascending
        # and first if descending", which reversing the whole key achieves.
        return (value is None, value if value is not None else "")

    return sorted(resources, key=key, reverse=descending)
# -- sorting-end --


def save_record(record):
    """Persist *record*, raising UniquenessException if its userName is already taken."""
    if not record.get("id"):
        record["id"] = str(uuid4())
    for existing in records.values():
        if (
            existing["id"] != record["id"]
            and existing["user_name"] == record["user_name"]
        ):
            raise UniquenessException(
                detail=f"userName {record['user_name']!r} is already taken"
            )
    now = datetime.now(timezone.utc)
    record.setdefault("created_at", now)
    record["updated_at"] = now
    records[record["id"]] = record


def delete_record(record_id):
    """Remove the record identified by *record_id*."""
    del records[record_id]


# The root query needs a second resource type to gather. These guides do not
# implement the ``/Groups`` endpoints, so groups are read-only fixtures.
group_records = {
    "6c8a2e1f": {"id": "6c8a2e1f", "display_name": "Administrators"},
    "b3f1d049": {"id": "b3f1d049", "display_name": "Auditors"},
}


def list_group_records():
    """Return every stored group record."""
    return list(group_records.values())

Mapping application data to SCIM

scim2-models suppose that your application storage layer has its own internal model and does not use SCIM models internally. You need mapping helpers that convert between your application representation and the SCIM resource exposed over HTTP — here User, but the same approach works for Group or any other resource type.

Example of serialization and deserialization between scim2 and custom model representation

def to_scim_user(record, location=None):
    """Convert an application record into a SCIM User resource.

    :param record: The application record.
    :param location: Canonical URL of the resource, set in :attr:`~scim2_models.Meta.location`.
    """
    return User(
        id=record["id"],
        user_name=record["user_name"],
        display_name=record.get("display_name"),
        active=record.get("active", True),
        emails=[User.Emails(value=record["email"])] if record.get("email") else None,
        meta=Meta(
            resource_type="User",
            version=make_etag(record),
            created=record["created_at"],
            last_modified=record["updated_at"],
            location=location,
        ),
    )


def from_scim_user(scim_user):
    """Convert a validated SCIM payload into the application shape."""
    return {
        "id": scim_user.id,
        "user_name": scim_user.user_name,
        "display_name": scim_user.display_name,
        "active": True if scim_user.active is None else scim_user.active,
        "email": scim_user.emails[0].value if scim_user.emails else None,
    }


def make_etag(record):
    """Compute a weak ETag from a record's content."""
    digest = hashlib.sha256(str(sorted(record.items())).encode()).hexdigest()[:16]
    return f'W/"{digest}"'


def to_scim_group(record):
    """Convert an application group record into a SCIM Group resource.

    ``meta.location`` is left out, as these guides expose no ``/Groups``
    endpoint to point it at.

    :param record: The application group record.
    """
    return Group(
        id=record["id"],
        display_name=record["display_name"],
        meta=Meta(resource_type="Group"),
    )

This separation keeps the HTTP layer simple. The views work with SCIM resources, while the rest of the application can keep its own representation.

Ordering and paging collections

A collection endpoint answers the sortBy, sortOrder, startIndex and count parameters of RFC7644 §3.4.2. Naming the resource type the endpoint serves, with SearchRequest[User], resolves sort_by against that model, so the helper below reads Path.field_name instead of the attribute name a client spelled.

RFC7644 §3.4.2.3 decides the order in three ways the helper follows: a string attribute is compared without its case unless it is annotated CaseExact.true; a multi-valued attribute is compared on the value of its primary entry, or the first one; and a resource with no value for the attribute comes last when ascending, first when descending.

Ordering a collection

def page_of(resources, req):
    """Return the total count and the page a query asks for.

    Sorting comes before paging, so a page holds the same resources whatever
    the order asked for. A page never exceeds ``MAX_RESULTS`` entries, which is
    the bound the :class:`~scim2_models.ServiceProviderConfig` advertises.

    :param resources: The SCIM resources to answer from.
    :param req: The parsed query.
    :return: A ``(total, page)`` tuple.
    """
    if req.sort_by:
        resources = sort_resources(resources, req.sort_by, req.sort_order)

    start = req.start_index_0 or 0
    limit = start + MAX_RESULTS
    stop = req.stop_index_0
    stop = limit if stop is None else min(stop, limit)
    return len(resources), resources[start:stop]


def sort_resources(resources, sort_by, sort_order=None):
    """Order resources by an attribute, per :rfc:`RFC7644 §3.4.2.3 <7644#section-3.4.2.3>`.

    :param resources: The SCIM resources to order.
    :param sort_by: The ``sortBy`` query parameter, resolved by the request it
        came from, which names the resource type the endpoint serves.
    :param sort_order: The ``sortOrder`` query parameter, ascending by default.
    :raises InvalidPathException: If the attribute is unknown.
    """
    if sort_by.field_name is None:
        raise InvalidPathException(
            path=str(sort_by), detail=f"Cannot sort on {sort_by!r}"
        )

    # "String type attributes are case insensitive by default, unless the
    # attribute type is defined as a case-exact string."
    case_exact = sort_by.model.get_field_annotation(sort_by.field_name, CaseExact)
    descending = sort_order == SearchRequest.SortOrder.descending

    def key(resource):
        value = sort_by.get(resource, strict=False)
        if isinstance(value, list):
            # "resources are sorted by the value of the primary attribute, if
            # any, or else the first value in the list, if any."
            primary = next((each for each in value if each.primary), None)
            entry = primary or (value[0] if value else None)
            value = entry.value if entry else None
        if isinstance(value, str) and not case_exact:
            value = value.casefold()
        # "if there is no data for the specified sortBy value, they are sorted
        # via the sortOrder parameter, i.e., they are ordered last if ascending
        # and first if descending", which reversing the whole key achieves.
        return (value is None, value if value is not None else "")

    return sorted(resources, key=key, reverse=descending)

Sorting comes before paging, so a page holds the same resources whatever the order asked for, and a page never exceeds the maxResults the ServiceProviderConfig advertises. Both are what page_of applies, and every collection endpoint of the guides goes through it.

Server discovery

SCIM clients discover the server capabilities by querying three read-only endpoints: /Schemas, /ResourceTypes and /ServiceProviderConfig (RFC 7644 §4). The helpers below build Schema and ResourceType objects from the resource models your server exposes, and define a ServiceProviderConfig describing the server’s capabilities.

Server discovery helpers

RESOURCE_MODELS = [User]


def get_schemas():
    """Return every :class:`~scim2_models.Schema` the server exposes."""
    return [model.to_schema() for model in RESOURCE_MODELS]


def get_schema(schema_id):
    """Return the :class:`~scim2_models.Schema` matching *schema_id*, or raise KeyError."""
    for model in RESOURCE_MODELS:
        schema = model.to_schema()
        if schema.id == schema_id:
            return schema
    raise KeyError(schema_id)


def get_resource_types():
    """Return every :class:`~scim2_models.ResourceType` the server exposes."""
    return [ResourceType.from_resource(model) for model in RESOURCE_MODELS]


def get_resource_type(resource_type_id):
    """Return the :class:`~scim2_models.ResourceType` matching *resource_type_id*, or raise KeyError."""
    for model in RESOURCE_MODELS:
        rt = ResourceType.from_resource(model)
        if rt.id == resource_type_id:
            return rt
    raise KeyError(resource_type_id)


service_provider_config = ServiceProviderConfig(
    patch=Patch(supported=True),
    bulk=Bulk(supported=False, max_operations=0, max_payload_size=0),
    filter=Filter(supported=False, max_results=0),
    change_password=ChangePassword(supported=False),
    sort=Sort(supported=True),
    etag=ETag(supported=True),
    authentication_schemes=[
        AuthenticationScheme(
            type=AuthenticationScheme.Type.httpbasic,
            name="HTTP Basic",
            description="Authentication via HTTP Basic",
        ),
    ],
)

Web frameworks

Those sections show how to process incoming SCIM HTTP requests, and which response to produce.

• Flask
• Django
• FastAPI