FastAPI#

This guide shows a minimal SCIM server built with FastAPI and scim2_models. It shows how to:

validate incoming SCIM payloads with the right Context;
serialize resources and collections as SCIM responses;
expose responses with the application/scim+json media type;
convert validation errors into SCIM Error payloads.

The example uses User as a concrete resource type, but the same pattern applies to any other resource such as Group. The storage and mapping layers are defined in Shared helpers for a SCIM server, and shared by every integration example. The complete runnable file is available in the Complete example section.

pip install fastapi uvicorn scim2-models

Application setup#

Start with a FastAPI application and an APIRouter prefixed with /scim/v2. SCIMResponse is a thin Response subclass that sets the application/scim+json content type and automatically extracts the ETag header from meta.version when the response body contains it.

app = FastAPI()


class SCIMResponse(JSONResponse):
    """SCIM JSON response that auto-extracts the ``ETag`` from ``meta.version``."""

    media_type = "application/scim+json"

    def __init__(self, content: Any = None, **kwargs: Any) -> None:
        super().__init__(content, **kwargs)
        if meta := (content or {}).get("meta", {}):
            if version := meta.get("version"):
                self.headers["ETag"] = version


router = APIRouter(prefix="/scim/v2", default_response_class=SCIMResponse)



# -- provider-middleware-start --
@app.middleware("http")
async def scim_provider(request: Request, call_next):
    """Validate the payloads under the provider, which knows the resource type of each endpoint."""
    with provider:
        return await call_next(request)
# -- provider-middleware-end --


def resource_location(request, app_record):
    """Return the canonical URL for a user record."""
    return str(request.url_for("get_user", user_id=app_record["id"]))

Endpoints#

The routes of this section serve /Users, but the same structure applies to any resource type: replace the mapping helpers, the model class, and the URL prefix to expose /Groups or any other collection. Write endpoints use SCIMValidator to let FastAPI parse and validate the request body with the correct SCIM Context automatically. Read endpoints still build responses explicitly because they need to forward attributes / excludedAttributes query parameters.

GET /Users/<id>#

Parse query parameters with ResponseParameters, convert the native record to a SCIM resource with a mapping helper, then serialize with RESOURCE_QUERY_RESPONSE, forwarding req.attributes and req.excluded_attributes so the response only includes the requested fields.

@router.get("/Users/{user_id}")
async def get_user(
    request: Request,
    req: Annotated[ResponseParameters, Query()],
    app_record: dict = Depends(resolve_user),
):
    """Return one SCIM user."""
    scim_user = to_scim_user(app_record, resource_location(request, app_record))
    etag = make_etag(app_record)
    if_none_match = request.headers.get("If-None-Match")
    if if_none_match and etag in [t.strip() for t in if_none_match.split(",")]:
        return Response(status_code=HTTPStatus.NOT_MODIFIED)
    return SCIMResponse(
        scim_user.model_dump(
            scim_ctx=Context.RESOURCE_QUERY_RESPONSE,
            response_parameters=req,
        ),
    )

DELETE /Users/<id>#

Remove the record from the store and return an empty 204 response. No SCIM serialization is needed.

@router.delete("/Users/{user_id}")
async def delete_user(request: Request, app_record: dict = Depends(resolve_user)):
    """Delete an existing user."""
    check_etag(app_record, request)
    delete_record(app_record["id"])
    return Response(status_code=HTTPStatus.NO_CONTENT)

PATCH /Users/<id>#

The patch payload is validated through SCIMValidator with RESOURCE_PATCH_REQUEST. Apply it to a SCIM conversion of the native record with patch(), convert back to native and persist, then serialize the result with RESOURCE_PATCH_RESPONSE. PatchOp is generic and works with any resource type.

@router.patch("/Users/{user_id}")
async def patch_user(
    request: Request,
    patch: PatchRequestContext[PatchOp[User]],
    req: Annotated[ResponseParameters, Query()],
    app_record: dict = Depends(resolve_user),
):
    """Apply a SCIM PatchOp to an existing user."""
    check_etag(app_record, request)
    scim_user = to_scim_user(app_record, resource_location(request, app_record))
    patch.patch(scim_user)

    updated_record = from_scim_user(scim_user)
    save_record(updated_record)

    response_user = to_scim_user(updated_record, resource_location(request, updated_record))
    return SCIMResponse(
        response_user.model_dump(
            scim_ctx=Context.RESOURCE_PATCH_RESPONSE,
            response_parameters=req,
        ),
    )

PUT /Users/<id>#

The full replacement payload is validated through SCIMValidator with RESOURCE_REPLACEMENT_REQUEST, then call replace() to verify that immutable attributes have not been modified.

@router.put("/Users/{user_id}")
async def replace_user(
    request: Request,
    replacement: ReplacementRequestContext[User],
    req: Annotated[ResponseParameters, Query()],
    app_record: dict = Depends(resolve_user),
):
    """Replace an existing user with a full SCIM resource."""
    check_etag(app_record, request)
    existing_user = to_scim_user(app_record, resource_location(request, app_record))
    replacement.replace(existing_user)

    updated_record = from_scim_user(replacement)
    save_record(updated_record)

    response_user = to_scim_user(updated_record, resource_location(request, updated_record))
    return SCIMResponse(
        response_user.model_dump(
            scim_ctx=Context.RESOURCE_REPLACEMENT_RESPONSE,
            response_parameters=req,
        ),
    )

GET /Users#

Parse the query parameters with SearchRequest[User], keep the resources the filter accepts as described in Filtering, order and page them as described in Ordering and paging collections, then wrap the page in a ListResponse serialized with RESOURCE_QUERY_RESPONSE. Pass req as response_parameters to model_dump_json() so that the attributes and excludedAttributes query parameters are applied to each embedded resource.

def users_response(request, req, scim_ctx):
    """Return one page of users as a serialized SCIM ListResponse.

    A query applies to the SCIM representation rather than to the stored
    records, so the store is mapped before it is filtered, ordered and
    paginated.

    :param request: The incoming request, used to build resource locations.
    :param req: The parsed query, whichever verb carried it.
    :param scim_ctx: The context to serialize the response in.
    """
    users = [
        to_scim_user(record, resource_location(request, record))
        for record in list_records()
    ]
    if req.filter:
        users = [user for user in users if req.filter.match(user)]
    total, page = page_of(users, req)
    response = ListResponse[User](
        total_results=total,
        start_index=req.start_index or 1,
        items_per_page=len(page),
        resources=page,
    )
    return SCIMResponse(
        response.model_dump(
            scim_ctx=scim_ctx,
            response_parameters=req,
        ),
    )


@router.get("/Users")
async def list_users(
    request: Request, req: Annotated[SearchRequest[User], Query()]
):
    """Return one page of users as a SCIM ListResponse."""
    return users_response(request, req, Context.RESOURCE_QUERY_RESPONSE)

POST /Users/.search#

SCIM lets a client send the same query in a body rather than on the URL, by appending /.search to any endpoint. The parameters are the ones of GET /Users, so the two verbs differ only in the parsing: validate the body with SEARCH_REQUEST and serialize the response with SEARCH_RESPONSE.

@router.post("/Users/.search")
async def search_users(
    request: Request, req: SearchRequestContext[SearchRequest[User]]
):
    """Answer the same query as GET /Users, with the parameters in the body."""
    return users_response(request, req, Context.SEARCH_RESPONSE)

POST /.search#

The same extension on the server root queries every resource type the server serves. The endpoint gathers each type it serves and answers with a ListResponse[User | Group].

Binding the request to that same union checks the filter against both models. An attribute only one of them declares is valid, and evaluates to false on the other, as §3.4.2.1 requires: “for filtered attributes that are not part of a particular resource type, the service provider SHALL treat the attribute as if there is no attribute value”. So userName pr keeps the users, and the request refuses an attribute neither model declares.

This section exposes no /Groups endpoints; the groups the root query gathers are read-only fixtures, enough to show a heterogeneous collection.

A server can refuse a root query whose result set would be too large with a tooMany error.

@router.post("/.search")
async def search_root(
    request: Request, req: SearchRequestContext[SearchRequest[User | Group]]
):
    """Query every resource type the server serves.

    :rfc:`RFC7644 §3.4.2.1 <7644#section-3.4.2.1>` has a root query cover them
    all, so the request is parameterised with a union: the filter is checked
    against both models, and an attribute only one of them declares evaluates
    to false on the other. These guides expose no ``/Groups`` endpoint, yet the
    root query gathers groups all the same.
    """
    resources = [
        to_scim_user(record, resource_location(request, record))
        for record in list_records()
    ]
    resources += [to_scim_group(record) for record in list_group_records()]
    if req.filter:
        resources = [
            resource for resource in resources if req.filter.match(resource)
        ]
    total, page = page_of(resources, req)
    response = ListResponse[User | Group](
        total_results=total,
        start_index=req.start_index or 1,
        items_per_page=len(page),
        resources=page,
    )
    return SCIMResponse(
        response.model_dump(
            scim_ctx=Context.SEARCH_RESPONSE,
            response_parameters=req,
        ),
    )

POST /Users#

The creation payload is validated through SCIMValidator with RESOURCE_CREATION_REQUEST. Convert to native and persist, then serialize the created resource with RESOURCE_CREATION_RESPONSE.

@router.post("/Users", status_code=HTTPStatus.CREATED)
async def create_user(
    request: Request,
    request_user: CreationRequestContext[User],
    req: Annotated[ResponseParameters, Query()],
):
    """Validate a SCIM creation payload and store the new user."""
    app_record = from_scim_user(request_user)
    save_record(app_record)

    response_user = to_scim_user(app_record, resource_location(request, app_record))
    return SCIMResponse(
        response_user.model_dump(
            scim_ctx=Context.RESOURCE_CREATION_RESPONSE,
            response_parameters=req,
        ),
        status_code=HTTPStatus.CREATED,
    )

POST /Bulk#

The job is validated through BulkRequestContext, which applies BULK_REQUEST to the annotated model, and the outcome is serialized with BULK_RESPONSE. The route closes over its request to build each location.

FastAPI validates the job before it calls the route, so the provider is opened by a middleware. Each operation is then read as the resource type its path targets.

@app.middleware("http")
async def scim_provider(request: Request, call_next):
    """Validate the payloads under the provider, which knows the resource type of each endpoint."""
    with provider:
        return await call_next(request)

A job that exceeds maxOperations raises SCIMException, which the handler registered for it turns into a 413. See Bulk jobs for what the executor does with each operation.

@router.post("/Bulk")
async def bulk(
    request: Request, bulk_request: BulkRequestContext[BulkRequest[User]]
):
    """Apply a bulk job and answer one result per operation."""
    bulk_response = execute_bulk(
        bulk_request, lambda record: resource_location(request, record)
    )
    return SCIMResponse(bulk_response.model_dump(scim_ctx=Context.BULK_RESPONSE))

Resource versioning (ETags)#

SCIM supports resource versioning through HTTP ETags. check_etag reads the If-Match header from the incoming request, compares it against the record’s ETag and raises an HTTPException on mismatch. make_etag computes a weak ETag from each record and populates version.

def check_etag(record, request: Request):
    """Compare the record's ETag against the ``If-Match`` request header.

    :param record: The application record.
    :param request: The incoming request.
    :raises ~fastapi.HTTPException: If the header is present and does not match.
    """
    if_match = request.headers.get("If-Match")
    if not if_match:
        return
    if if_match.strip() == "*":
        return
    etag = make_etag(record)
    tags = [t.strip() for t in if_match.split(",")]
    if etag not in tags:
        raise HTTPException(status_code=412, detail="ETag mismatch")

On GET single-resource responses, the If-None-Match request header is checked manually to return a 304 Not Modified when the client already has the current version.

On write operations (PUT, PATCH, DELETE), the If-Match header is checked before processing. If the client’s ETag does not match, a 412 Precondition Failed SCIM error is returned.

SCIMResponse automatically extracts meta.version from the serialized body and sets the ETag response header, so endpoints do not need to set it manually.

Tip

When the application uses SQLAlchemy, the built-in version counter can serve as ETag value directly, removing the need for a manual hash.

Discovery endpoints#

SCIM defines three read-only endpoints that let clients discover the server’s capabilities and the resources it exposes. The shared discovery helpers that build Schema, ResourceType and ServiceProviderConfig objects are defined in Shared helpers for a SCIM server.

GET /Schemas and GET /Schemas/<id>#

Return all Schema objects or look one up by its URI. The ScimProvider derives them from the resource models it holds. The collection endpoint parses pagination parameters with SearchRequest, following the same pattern as GET /Users.

@router.get("/Schemas")
async def list_schemas(req: Annotated[SearchRequest[Schema], Query()]):
    """Return one page of SCIM schemas the server exposes."""
    total, page = page_of(provider.schemas, req)
    response = ListResponse[Schema](
        total_results=total,
        start_index=req.start_index or 1,
        items_per_page=len(page),
        resources=page,
    )
    return SCIMResponse(
        response.model_dump(scim_ctx=Context.RESOURCE_QUERY_RESPONSE),
    )


@router.get("/Schemas/{schema_id:path}")
async def get_schema_by_id(schema_id: str):
    """Return one SCIM schema by its URI identifier."""
    try:
        schema = get_schema(schema_id)
    except KeyError:
        scim_error = Error(status=404, detail=f"Schema {schema_id!r} not found")
        return SCIMResponse(scim_error.model_dump(), status_code=HTTPStatus.NOT_FOUND)
    return SCIMResponse(
        schema.model_dump(scim_ctx=Context.RESOURCE_QUERY_RESPONSE),
    )

GET /ResourceTypes and GET /ResourceTypes/<id>#

Return all ResourceType objects or look one up by its identifier. The ScimProvider derives them from the resource models it holds. The collection endpoint parses pagination parameters with SearchRequest, following the same pattern as GET /Users.

@router.get("/ResourceTypes")
async def list_resource_types(req: Annotated[SearchRequest[ResourceType], Query()]):
    """Return one page of SCIM resource types the server exposes."""
    total, page = page_of(provider.resource_types, req)
    response = ListResponse[ResourceType](
        total_results=total,
        start_index=req.start_index or 1,
        items_per_page=len(page),
        resources=page,
    )
    return SCIMResponse(
        response.model_dump(scim_ctx=Context.RESOURCE_QUERY_RESPONSE),
    )


@router.get("/ResourceTypes/{resource_type_id}")
async def get_resource_type_by_id(resource_type_id: str):
    """Return one SCIM resource type by its identifier."""
    try:
        rt = get_resource_type(resource_type_id)
    except KeyError:
        scim_error = Error(
            status=404, detail=f"ResourceType {resource_type_id!r} not found"
        )
        return SCIMResponse(scim_error.model_dump(), status_code=HTTPStatus.NOT_FOUND)
    return SCIMResponse(
        rt.model_dump(scim_ctx=Context.RESOURCE_QUERY_RESPONSE),
    )

GET /ServiceProviderConfig#

Return the ServiceProviderConfig singleton that describes the features the server supports (patch, bulk, filtering, etc.).

@router.get("/ServiceProviderConfig")
async def get_service_provider_config() -> QueryResponseContext[
    ServiceProviderConfig
]:
    """Return the SCIM service provider configuration."""
    return provider.config

Idiomatic type annotations#

The write endpoints of the Endpoints section use the context type aliases provided by scim2_models. *RequestContext aliases wrap SCIMValidator (input validation), *ResponseContext aliases wrap SCIMSerializer (output serialization):

from scim2_models import CreationRequestContext, CreationResponseContext, User

@router.post("/Users", status_code=201)
async def create_user(
    user: CreationRequestContext[User],
) -> CreationResponseContext[User]:
    app_record = from_scim_user(user)
    save_record(app_record)
    return to_scim_user(app_record, ...)

Available aliases: CreationRequestContext / CreationResponseContext, QueryRequestContext / QueryResponseContext, ReplacementRequestContext / ReplacementResponseContext, SearchRequestContext / SearchResponseContext, and PatchRequestContext / PatchResponseContext.

These aliases are pure Pydantic and carry no dependency on FastAPI: they work with any framework that respects typing.Annotated metadata.

*ResponseContext aliases do not support the attributes / excludedAttributes query parameters. Forwarding those parameters takes the explicit model_dump_json() of the Endpoints section instead.

Complete example#

from http import HTTPStatus
from typing import Annotated
from typing import Any

from fastapi import APIRouter
from fastapi import Depends
from fastapi import FastAPI
from fastapi import HTTPException
from fastapi import Query
from fastapi import Request
from fastapi import Response
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from pydantic import ValidationError

from scim2_models import BulkRequest
from scim2_models import BulkRequestContext
from scim2_models import Context
from scim2_models import CreationRequestContext
from scim2_models import Error
from scim2_models import Group
from scim2_models import ListResponse
from scim2_models import PatchOp
from scim2_models import PatchRequestContext
from scim2_models import QueryResponseContext
from scim2_models import ReplacementRequestContext
from scim2_models import ResourceType
from scim2_models import ResponseParameters
from scim2_models import Schema
from scim2_models import SCIMException
from scim2_models import ServiceProviderConfig
from scim2_models import SearchRequest
from scim2_models import SearchRequestContext
from scim2_models import User

from .integrations import delete_record
from .integrations import execute_bulk
from .integrations import from_scim_user
from .integrations import get_record
from .integrations import get_resource_type
from .integrations import get_schema
from .integrations import list_group_records
from .integrations import list_records
from .integrations import page_of
from .integrations import make_etag
from .integrations import save_record
from .integrations import provider
from .integrations import to_scim_group
from .integrations import to_scim_user

# -- setup-start --
app = FastAPI()


class SCIMResponse(JSONResponse):
    """SCIM JSON response that auto-extracts the ``ETag`` from ``meta.version``."""

    media_type = "application/scim+json"

    def __init__(self, content: Any = None, **kwargs: Any) -> None:
        super().__init__(content, **kwargs)
        if meta := (content or {}).get("meta", {}):
            if version := meta.get("version"):
                self.headers["ETag"] = version


router = APIRouter(prefix="/scim/v2", default_response_class=SCIMResponse)



# -- provider-middleware-start --
@app.middleware("http")
async def scim_provider(request: Request, call_next):
    """Validate the payloads under the provider, which knows the resource type of each endpoint."""
    with provider:
        return await call_next(request)
# -- provider-middleware-end --


def resource_location(request, app_record):
    """Return the canonical URL for a user record."""
    return str(request.url_for("get_user", user_id=app_record["id"]))
# -- setup-end --


# -- etag-start --
def check_etag(record, request: Request):
    """Compare the record's ETag against the ``If-Match`` request header.

    :param record: The application record.
    :param request: The incoming request.
    :raises ~fastapi.HTTPException: If the header is present and does not match.
    """
    if_match = request.headers.get("If-Match")
    if not if_match:
        return
    if if_match.strip() == "*":
        return
    etag = make_etag(record)
    tags = [t.strip() for t in if_match.split(",")]
    if etag not in tags:
        raise HTTPException(status_code=412, detail="ETag mismatch")
# -- etag-end --


# -- refinements-start --
# -- dependency-start --
def resolve_user(user_id: str):
    """Resolve a user identifier to an application record."""
    try:
        return get_record(user_id)
    except KeyError:
        raise HTTPException(status_code=HTTPStatus.NOT_FOUND)
# -- dependency-end --


# -- error-handlers-start --
@app.exception_handler(ValidationError)
@app.exception_handler(RequestValidationError)
async def handle_validation_error(request, error):
    """Turn Pydantic validation errors into SCIM error responses."""
    scim_error = Error.from_validation_error(error.errors()[0])
    return SCIMResponse(scim_error.model_dump(), status_code=scim_error.status)


@app.exception_handler(HTTPException)
async def handle_http_exception(request, error):
    """Turn HTTP exceptions into SCIM error responses."""
    scim_error = Error(status=error.status_code, detail=error.detail or "")
    return SCIMResponse(scim_error.model_dump(), status_code=error.status_code)


@app.exception_handler(SCIMException)
async def handle_scim_error(request, error):
    """Turn SCIM exceptions into SCIM error responses."""
    scim_error = error.to_error()
    return SCIMResponse(scim_error.model_dump(), status_code=scim_error.status)
# -- error-handlers-end --
# -- refinements-end --


# -- endpoints-start --
# -- single-resource-start --
# -- get-user-start --
@router.get("/Users/{user_id}")
async def get_user(
    request: Request,
    req: Annotated[ResponseParameters, Query()],
    app_record: dict = Depends(resolve_user),
):
    """Return one SCIM user."""
    scim_user = to_scim_user(app_record, resource_location(request, app_record))
    etag = make_etag(app_record)
    if_none_match = request.headers.get("If-None-Match")
    if if_none_match and etag in [t.strip() for t in if_none_match.split(",")]:
        return Response(status_code=HTTPStatus.NOT_MODIFIED)
    return SCIMResponse(
        scim_user.model_dump(
            scim_ctx=Context.RESOURCE_QUERY_RESPONSE,
            response_parameters=req,
        ),
    )
# -- get-user-end --


# -- patch-user-start --
@router.patch("/Users/{user_id}")
async def patch_user(
    request: Request,
    patch: PatchRequestContext[PatchOp[User]],
    req: Annotated[ResponseParameters, Query()],
    app_record: dict = Depends(resolve_user),
):
    """Apply a SCIM PatchOp to an existing user."""
    check_etag(app_record, request)
    scim_user = to_scim_user(app_record, resource_location(request, app_record))
    patch.patch(scim_user)

    updated_record = from_scim_user(scim_user)
    save_record(updated_record)

    response_user = to_scim_user(updated_record, resource_location(request, updated_record))
    return SCIMResponse(
        response_user.model_dump(
            scim_ctx=Context.RESOURCE_PATCH_RESPONSE,
            response_parameters=req,
        ),
    )
# -- patch-user-end --


# -- put-user-start --
@router.put("/Users/{user_id}")
async def replace_user(
    request: Request,
    replacement: ReplacementRequestContext[User],
    req: Annotated[ResponseParameters, Query()],
    app_record: dict = Depends(resolve_user),
):
    """Replace an existing user with a full SCIM resource."""
    check_etag(app_record, request)
    existing_user = to_scim_user(app_record, resource_location(request, app_record))
    replacement.replace(existing_user)

    updated_record = from_scim_user(replacement)
    save_record(updated_record)

    response_user = to_scim_user(updated_record, resource_location(request, updated_record))
    return SCIMResponse(
        response_user.model_dump(
            scim_ctx=Context.RESOURCE_REPLACEMENT_RESPONSE,
            response_parameters=req,
        ),
    )
# -- put-user-end --


# -- delete-user-start --
@router.delete("/Users/{user_id}")
async def delete_user(request: Request, app_record: dict = Depends(resolve_user)):
    """Delete an existing user."""
    check_etag(app_record, request)
    delete_record(app_record["id"])
    return Response(status_code=HTTPStatus.NO_CONTENT)
# -- delete-user-end --
# -- single-resource-end --


# -- collection-start --
# -- list-users-start --
def users_response(request, req, scim_ctx):
    """Return one page of users as a serialized SCIM ListResponse.

    A query applies to the SCIM representation rather than to the stored
    records, so the store is mapped before it is filtered, ordered and
    paginated.

    :param request: The incoming request, used to build resource locations.
    :param req: The parsed query, whichever verb carried it.
    :param scim_ctx: The context to serialize the response in.
    """
    users = [
        to_scim_user(record, resource_location(request, record))
        for record in list_records()
    ]
    if req.filter:
        users = [user for user in users if req.filter.match(user)]
    total, page = page_of(users, req)
    response = ListResponse[User](
        total_results=total,
        start_index=req.start_index or 1,
        items_per_page=len(page),
        resources=page,
    )
    return SCIMResponse(
        response.model_dump(
            scim_ctx=scim_ctx,
            response_parameters=req,
        ),
    )


@router.get("/Users")
async def list_users(
    request: Request, req: Annotated[SearchRequest[User], Query()]
):
    """Return one page of users as a SCIM ListResponse."""
    return users_response(request, req, Context.RESOURCE_QUERY_RESPONSE)
# -- list-users-end --


# -- search-users-start --
@router.post("/Users/.search")
async def search_users(
    request: Request, req: SearchRequestContext[SearchRequest[User]]
):
    """Answer the same query as GET /Users, with the parameters in the body."""
    return users_response(request, req, Context.SEARCH_RESPONSE)
# -- search-users-end --


# -- search-root-start --
@router.post("/.search")
async def search_root(
    request: Request, req: SearchRequestContext[SearchRequest[User | Group]]
):
    """Query every resource type the server serves.

    :rfc:`RFC7644 §3.4.2.1 <7644#section-3.4.2.1>` has a root query cover them
    all, so the request is parameterised with a union: the filter is checked
    against both models, and an attribute only one of them declares evaluates
    to false on the other. These guides expose no ``/Groups`` endpoint, yet the
    root query gathers groups all the same.
    """
    resources = [
        to_scim_user(record, resource_location(request, record))
        for record in list_records()
    ]
    resources += [to_scim_group(record) for record in list_group_records()]
    if req.filter:
        resources = [
            resource for resource in resources if req.filter.match(resource)
        ]
    total, page = page_of(resources, req)
    response = ListResponse[User | Group](
        total_results=total,
        start_index=req.start_index or 1,
        items_per_page=len(page),
        resources=page,
    )
    return SCIMResponse(
        response.model_dump(
            scim_ctx=Context.SEARCH_RESPONSE,
            response_parameters=req,
        ),
    )
# -- search-root-end --


# -- create-user-start --
@router.post("/Users", status_code=HTTPStatus.CREATED)
async def create_user(
    request: Request,
    request_user: CreationRequestContext[User],
    req: Annotated[ResponseParameters, Query()],
):
    """Validate a SCIM creation payload and store the new user."""
    app_record = from_scim_user(request_user)
    save_record(app_record)

    response_user = to_scim_user(app_record, resource_location(request, app_record))
    return SCIMResponse(
        response_user.model_dump(
            scim_ctx=Context.RESOURCE_CREATION_RESPONSE,
            response_parameters=req,
        ),
        status_code=HTTPStatus.CREATED,
    )
# -- create-user-end --
# -- collection-end --


# -- bulk-start --
@router.post("/Bulk")
async def bulk(
    request: Request, bulk_request: BulkRequestContext[BulkRequest[User]]
):
    """Apply a bulk job and answer one result per operation."""
    bulk_response = execute_bulk(
        bulk_request, lambda record: resource_location(request, record)
    )
    return SCIMResponse(bulk_response.model_dump(scim_ctx=Context.BULK_RESPONSE))
# -- bulk-end --


# -- discovery-start --
# -- schemas-start --
@router.get("/Schemas")
async def list_schemas(req: Annotated[SearchRequest[Schema], Query()]):
    """Return one page of SCIM schemas the server exposes."""
    total, page = page_of(provider.schemas, req)
    response = ListResponse[Schema](
        total_results=total,
        start_index=req.start_index or 1,
        items_per_page=len(page),
        resources=page,
    )
    return SCIMResponse(
        response.model_dump(scim_ctx=Context.RESOURCE_QUERY_RESPONSE),
    )


@router.get("/Schemas/{schema_id:path}")
async def get_schema_by_id(schema_id: str):
    """Return one SCIM schema by its URI identifier."""
    try:
        schema = get_schema(schema_id)
    except KeyError:
        scim_error = Error(status=404, detail=f"Schema {schema_id!r} not found")
        return SCIMResponse(scim_error.model_dump(), status_code=HTTPStatus.NOT_FOUND)
    return SCIMResponse(
        schema.model_dump(scim_ctx=Context.RESOURCE_QUERY_RESPONSE),
    )
# -- schemas-end --


# -- resource-types-start --
@router.get("/ResourceTypes")
async def list_resource_types(req: Annotated[SearchRequest[ResourceType], Query()]):
    """Return one page of SCIM resource types the server exposes."""
    total, page = page_of(provider.resource_types, req)
    response = ListResponse[ResourceType](
        total_results=total,
        start_index=req.start_index or 1,
        items_per_page=len(page),
        resources=page,
    )
    return SCIMResponse(
        response.model_dump(scim_ctx=Context.RESOURCE_QUERY_RESPONSE),
    )


@router.get("/ResourceTypes/{resource_type_id}")
async def get_resource_type_by_id(resource_type_id: str):
    """Return one SCIM resource type by its identifier."""
    try:
        rt = get_resource_type(resource_type_id)
    except KeyError:
        scim_error = Error(
            status=404, detail=f"ResourceType {resource_type_id!r} not found"
        )
        return SCIMResponse(scim_error.model_dump(), status_code=HTTPStatus.NOT_FOUND)
    return SCIMResponse(
        rt.model_dump(scim_ctx=Context.RESOURCE_QUERY_RESPONSE),
    )
# -- resource-types-end --


# -- service-provider-config-start --
@router.get("/ServiceProviderConfig")
async def get_service_provider_config() -> QueryResponseContext[
    ServiceProviderConfig
]:
    """Return the SCIM service provider configuration."""
    return provider.config
# -- service-provider-config-end --
# -- discovery-end --

app.include_router(router)
# -- endpoints-end --

FastAPI#

Application setup#

Optional FastAPI refinements#

Dependencies#

Exception handlers#

Endpoints#

GET /Users/<id>#

DELETE /Users/<id>#

PATCH /Users/<id>#

PUT /Users/<id>#

GET /Users#

POST /Users/.search#

POST /.search#

POST /Users#

POST /Bulk#

Resource versioning (ETags)#

Discovery endpoints#

GET /Schemas and GET /Schemas/<id>#

GET /ResourceTypes and GET /ResourceTypes/<id>#

GET /ServiceProviderConfig#

Idiomatic type annotations#

Complete example#