Integrate a web framework#

Use this guide to serve SCIM from a web application built with a framework, such as Flask, Django or FastAPI. It assumes a storage, such as the SQLite storage of Write a storage. It does not cover authentication: Authenticate and authorize the clients does.

The integration turns the request of the framework into a ScimRequest. scim2-server does everything else: it routes the request, validates it, calls the storage, and returns a ScimResponse. The integration turns it into a response of the framework.

Build the handler#

The views of the framework call a handler. Build it once, when the application starts. An asynchronous framework, such as FastAPI, uses an AsyncScimHandler over an AsyncScimStorage:

>>> from scim2_server.handler import ScimHandler
>>> from scim2_server.memory import InMemoryStorage
>>> from scim2_server.service import ScimService
>>> from scim2_server.utils import load_default_provider

>>> service = ScimService(load_default_provider())
>>> handler = ScimHandler(service, InMemoryStorage())

>>> from scim2_server.handler import AsyncScimHandler
>>> from scim2_server.memory import AsyncInMemoryStorage
>>> from scim2_server.service import ScimService
>>> from scim2_server.utils import load_default_provider

>>> service = ScimService(load_default_provider())
>>> handler = AsyncScimHandler(service, AsyncInMemoryStorage())

The handler needs two objects:

  • the ScimService applies the SCIM rules to the resources of a ScimProvider. This provider describes the users and the groups of RFC 7643. Keep a reference to the service, because the error handler of Return the response uses it.

  • the storage reads and writes the resources. Replace the in-memory storage with the storage of the application.

Read the request#

The handler serves generic SCIM requests, independent of any framework. Turn each request of the framework into a ScimRequest, with these values:

  • method: the HTTP method;

  • base_url: the root URL of the SCIM endpoints, as the client sees it, such as https://example.com/scim/v2. The handler builds the meta.location of the resources from it;

  • path: the path of the request, relative to base_url, such as /Users/2819c223;

  • query: the query parameters, as a mapping;

  • headers: the headers, as a mapping or as pairs. The handler reads Content-Type, If-Match and If-None-Match;

  • body: the raw body, as bytes;

  • subject: the authenticated client, if any. Authenticate and authorize the clients sets it.

The following sketch builds the request with Flask and FastAPI, for SCIM endpoints served under /scim/v2:

def scim_request():
    return ScimRequest(
        request.method,
        request.url_root + "scim/v2",
        request.path.removeprefix("/scim/v2"),
        request.args,
        request.headers,
        request.get_data(),
    )

async def scim_request(request: Request):
    return ScimRequest(
        request.method,
        str(request.base_url) + "scim/v2",
        request.url.path.removeprefix("/scim/v2"),
        request.query_params,
        request.headers.items(),
        await request.body(),
    )

The handler raises a PayloadTooLargeException for a bulk request larger than the maxPayloadSize of the service. It does not protect the memory of the server, since the body is already read. Limit the size of the bodies with the framework, such as the MAX_CONTENT_LENGTH setting of Flask. max_body_size() gives the limit of a request.

Route the requests#

The framework chooses the view that serves a request. The SCIM requests must reach the handler, by one of two ways: a single route for every request, or one route per SCIM operation.

handle() serves any SCIM request. It raises a NotFoundException for an unknown path, and a MethodNotAllowedException for a method that the endpoint does not support. Route every request under the SCIM root to it:

METHODS = ["GET", "POST", "PUT", "PATCH", "DELETE"]


@scim.route("/", defaults={"path": ""}, methods=METHODS)
@scim.route("/<path:path>", methods=METHODS)
def serve(path):
    return to_response(handler.handle(scim_request()))

@scim.api_route("/{path:path}", methods=["GET", "POST", "PUT", "PATCH", "DELETE"])
async def serve(request: Request):
    return to_response(await handler.handle(await scim_request(request)))

An integration can also keep the routing of the framework, for the OpenAPI schema of FastAPI or for a decorator on a single route. It then registers one route per SCIM operation. Each view calls the handler method of its operation, such as query() for a read. A last route sends the other requests to handle(), and the handler raises a SCIM error for them:

@scim.get("/<endpoint>/<resource_id>")
def query(endpoint, resource_id):
    return to_response(handler.query(scim_request()))


@scim.post("/<endpoint>")
def create(endpoint):
    return to_response(handler.create(scim_request()))


# One view per SCIM operation, then:


@scim.route("/<path:path>", methods=METHODS)
def other(path):
    return to_response(handler.handle(scim_request()))

@scim.get("/{endpoint}/{resource_id}")
async def query(request: Request):
    return to_response(await handler.query(await scim_request(request)))


@scim.post("/{endpoint}")
async def create(request: Request):
    return to_response(await handler.create(await scim_request(request)))


# One view per SCIM operation, then:


@scim.api_route("/{path:path}", methods=METHODS)
async def other(request: Request):
    return to_response(await handler.handle(await scim_request(request)))

ROUTES lists the routes of the SCIM operations. Each Route has a method, a pattern such as /{endpoint}/{resource_id}, and an operation. An integration can register its views from it. Register the routes in the order of the list: the routes with a fixed path, such as /Schemas, come first.

The handler reads the path itself: the routes of the framework only select the view. A handler method refuses a request of another operation. Such a refusal reveals a route registered in the wrong order.

Return the response#

The handler returns generic SCIM responses, independent of any framework. Turn each ScimResponse into a response of the framework, with these values:

  • the status from status;

  • the headers from headers, which hold the Content-Type, and the ETag and Location when the response has them;

  • the body from body, serialized as JSON, or no body when it is None.

The handler does not return a response when a request fails: it raises a SCIMException. Catch it in an error handler of the framework, and turn it into a response with error_response(). Any other exception is a bug, and the framework answers it with a 500.

The following sketch catches the SCIM exceptions with Flask and FastAPI:

@app.errorhandler(SCIMException)
def handle_scim_exception(exception):
    return to_response(service.error_response(exception))

@app.exception_handler(SCIMException)
async def handle_scim_exception(request: Request, exception: SCIMException):
    return to_response(service.error_response(exception))

Serve SCIM without a framework#

The following applications put the previous sections together, with no framework: as a WSGI application with a synchronous handler, and as an ASGI application with an asynchronous one:

import json
from urllib.parse import parse_qsl
from wsgiref.util import application_uri

from scim2_models import SCIMException

from scim2_server.handler import ScimHandler
from scim2_server.memory import InMemoryStorage
from scim2_server.requests import ScimRequest
from scim2_server.responses import ScimResponse
from scim2_server.service import ScimService
from scim2_server.utils import load_default_provider

service = ScimService(load_default_provider())
handler = ScimHandler(service, InMemoryStorage())


def wsgi_to_scim_request(environ):
    """Turn a WSGI request into a SCIM request."""
    headers = {
        key.removeprefix("HTTP_").replace("_", "-"): value
        for key, value in environ.items()
        if key.startswith("HTTP_")
    }
    headers["Content-Type"] = environ.get("CONTENT_TYPE", "")
    return ScimRequest(
        method=environ["REQUEST_METHOD"],
        base_url=application_uri(environ),
        path=environ["PATH_INFO"],
        query=dict(parse_qsl(environ.get("QUERY_STRING", ""))),
        headers=headers,
        body=environ["wsgi.input"].read(int(environ.get("CONTENT_LENGTH") or 0)),
    )


def scim_to_wsgi_response(response: ScimResponse, start_response):
    """Send a SCIM response as a WSGI response."""
    body = b"" if response.body is None else json.dumps(response.body).encode()
    status = f"{response.status.value} {response.status.phrase}"
    start_response(status, list(response.headers.items()))
    return [body]


def application(environ, start_response):
    """Serve the SCIM endpoints at the root of the application."""
    try:
        response = handler.handle(wsgi_to_scim_request(environ))
    except SCIMException as exception:
        response = service.error_response(exception)
    return scim_to_wsgi_response(response, start_response)

import json
from urllib.parse import parse_qsl

from scim2_models import SCIMException

from scim2_server.handler import AsyncScimHandler
from scim2_server.memory import AsyncInMemoryStorage
from scim2_server.requests import ScimRequest
from scim2_server.responses import ScimResponse
from scim2_server.service import ScimService
from scim2_server.utils import load_default_provider

service = ScimService(load_default_provider())
handler = AsyncScimHandler(service, AsyncInMemoryStorage())


async def asgi_to_scim_request(scope, receive):
    """Turn an ASGI request into a SCIM request."""
    headers = [(name.decode(), value.decode()) for name, value in scope["headers"]]
    body = b""
    while True:
        message = await receive()
        body += message.get("body", b"")
        if not message.get("more_body"):
            break
    return ScimRequest(
        method=scope["method"],
        base_url=f"{scope['scheme']}://{dict(headers)['host']}{scope['root_path']}",
        path=scope["path"],
        query=dict(parse_qsl(scope["query_string"].decode())),
        headers=headers,
        body=body,
    )


async def scim_to_asgi_response(response: ScimResponse, send):
    """Send a SCIM response as an ASGI response."""
    body = b"" if response.body is None else json.dumps(response.body).encode()
    await send(
        {
            "type": "http.response.start",
            "status": response.status,
            "headers": [(k.encode(), v.encode()) for k, v in response.headers.items()],
        }
    )
    await send({"type": "http.response.body", "body": body})


async def application(scope, receive, send):
    """Serve the SCIM endpoints at the root of the application."""
    try:
        response = await handler.handle(await asgi_to_scim_request(scope, receive))
    except SCIMException as exception:
        response = service.error_response(exception)
    await scim_to_asgi_response(response, send)

scim2-server provides the complete version of these two applications: WSGIApplication and ASGIApplication. Their hooks observe or change each request, without a framework.