Write a storage#

Use this guide to keep the SCIM resources in a database table of their own. It assumes a ScimProvider that describes the service, such as the one of Overview. To serve the data that an application already keeps in its own tables, follow Serve an existing data model after this guide.

Decide what to store#

A storage can keep the resources anywhere: a SQL database through a driver or an ORM, a document database, an LDAP directory, or a remote API. The server only calls its methods. This guide keeps the resources in a SQLite database. The synchronous storage uses sqlite3, and the asynchronous one uses aiosqlite. A row holds a resource: its resource type, its identifier, the userName of a user, the other attributes as JSON, its dates and its version. A UNIQUE constraint refuses two users with the same userName, whatever the case of its ASCII letters:

TABLE = """
CREATE TABLE IF NOT EXISTS resources (
    resource_type TEXT NOT NULL,
    id TEXT NOT NULL,
    user_name TEXT COLLATE NOCASE,
    attributes TEXT NOT NULL,
    created TEXT NOT NULL,
    last_modified TEXT NOT NULL,
    version INTEGER NOT NULL,
    PRIMARY KEY (resource_type, id),
    UNIQUE (resource_type, user_name)
)
"""

Keep the resources#

ScimStorage declares the methods that the server calls to read and write the resources: get(), create(), update(), delete() and search(). The server validates the requests and applies them before it calls these methods: the storage only reads and writes resources.

Subclass ScimStorage, or AsyncScimStorage for an asynchronous server. The methods of an asynchronous storage are coroutines, and follow the same rules.

Pass the ScimProvider to the storage. Open the connection of an asynchronous storage on its first call, from the event loop that serves the requests:

class SQLiteStorage(ScimStorage):
    """Keep the resources in a table of a SQLite database."""

    def __init__(self, connection: sqlite3.Connection, provider: ScimProvider):
        self.connection = connection
        self.connection.row_factory = sqlite3.Row
        self.connection.execute(TABLE)
        self.provider = provider


class AsyncSQLiteStorage(AsyncScimStorage):
    """Keep the resources in a table of a SQLite database, with aiosqlite."""

    def __init__(self, path: str, provider: ScimProvider):
        self.path = path
        self.connection: aiosqlite.Connection | None = None
        self.provider = provider

    async def connect(self) -> aiosqlite.Connection:
        """Open the database on the first call, and create its table."""
        if self.connection is None:
            self.connection = await aiosqlite.connect(self.path)
            self.connection.row_factory = aiosqlite.Row
            await self.connection.execute(TABLE)
        return self.connection

    async def close(self) -> None:
        """Close the database, when the application stops."""
        if self.connection is not None:
            await self.connection.close()
            self.connection = None

Turn a row into a resource with the model that the provider gives for its resource type. Fill its id, and its Meta with resourceType, created, lastModified and version. The version is the ETag of the resource, so it must be a quoted entity tag, such as W/"3" (RFC 9110 §8.8.3). It must change on every write. The example counts the writes. Leave meta.location empty: the server builds it from its own URL. In the same way, a reference to another resource, such as members.$ref, can be relative to the SCIM root, such as Users/2819c223: the server returns its URL.

def row_to_resource(
    self, resource_type: ResourceType, row: sqlite3.Row
) -> Resource[Any]:
    model = self.provider.model_for(resource_type)
    resource = model.model_validate_json(row["attributes"])
    resource.id = row["id"]
    resource.meta = Meta(
        resource_type=resource_type.name,
        created=row["created"],
        last_modified=row["last_modified"],
        version=f'W/"{row["version"]}"',
    )
    return resource

The complete storage passes the checks of Check a storage. Each of the following sections adds one of its methods.

Read a resource#

get() returns a resource, from its resource type and its identifier. Raise NotFoundException when it does not exist:

def get(
    self,
    resource_type: ResourceType,
    resource_id: str,
    *,
    response_parameters: ResponseParameters[Any] | None = None,
) -> Resource[Any]:
    row = self.connection.execute(
        "SELECT * FROM resources WHERE resource_type = ? AND id = ?",
        (resource_type.name, resource_id),
    ).fetchone()
    if row is None:
        raise NotFoundException(
            detail=f"{resource_type.name} {resource_id} not found"
        )
    return self.row_to_resource(resource_type, row)

async def get(
    self,
    resource_type: ResourceType,
    resource_id: str,
    *,
    response_parameters: ResponseParameters[Any] | None = None,
) -> Resource[Any]:
    connection = await self.connect()
    cursor = await connection.execute(
        "SELECT * FROM resources WHERE resource_type = ? AND id = ?",
        (resource_type.name, resource_id),
    )
    row = await cursor.fetchone()
    if row is None:
        raise NotFoundException(
            detail=f"{resource_type.name} {resource_id} not found"
        )
    return self.row_to_resource(resource_type, row)

For a GET request, the server also passes response_parameters: the attributes and excludedAttributes of the request. The example ignores them and returns the whole resource. The server removes the other attributes from the response. A storage can use them to avoid loading what the response leaves out, such as the members of a group kept in another table. Path.iter_paths gives the attributes the response keeps:

kept = Path[model].iter_paths(
    attributes=response_parameters.attributes,
    excluded_attributes=response_parameters.excluded_attributes,
)

Without response_parameters, return the whole resource. The server needs it for a PUT, a PATCH or a DELETE request.

Create a resource#

create() stores a new resource. Choose its identifier, and start its version at 1:

def create(
    self, resource_type: ResourceType, resource: Resource[Any]
) -> Resource[Any]:
    resource_id = uuid.uuid4().hex
    now = datetime.now(UTC).isoformat()
    with self.writing():
        self.connection.execute(
            "INSERT INTO resources VALUES (?, ?, ?, ?, ?, ?, 1)",
            (
                resource_type.name,
                resource_id,
                getattr(resource, "user_name", None),
                resource.model_dump_json(),
                now,
                now,
            ),
        )
    return self.get(resource_type, resource_id)

async def create(
    self, resource_type: ResourceType, resource: Resource[Any]
) -> Resource[Any]:
    resource_id = uuid.uuid4().hex
    now = datetime.now(UTC).isoformat()
    async with self.writing() as connection:
        await connection.execute(
            "INSERT INTO resources VALUES (?, ?, ?, ?, ?, ?, 1)",
            (
                resource_type.name,
                resource_id,
                getattr(resource, "user_name", None),
                resource.model_dump_json(),
                now,
                now,
            ),
        )
    return await self.get(resource_type, resource_id)

Refuse a duplicate value#

In the default user resource, the userName of a user is unique, whatever its case. Compare the values in the form AttributeBinding.comparable returns, as the filters do. The example relies on COLLATE NOCASE, which only ignores the case of ASCII letters: Élise and élise remain two users. On a creation and on an update, raise UniquenessException when another user already has the value. The UNIQUE constraint of the table refuses such a write, even when two requests arrive at the same time. The example commits each write, and turns the refusal into the exception:

@contextmanager
def writing(self) -> Iterator[None]:
    """Commit a write, and refuse a userName another user already has."""
    try:
        with self.connection:
            yield
    except sqlite3.IntegrityError:
        raise UniquenessException from None

@asynccontextmanager
async def writing(self) -> AsyncIterator[aiosqlite.Connection]:
    """Commit a write, and refuse a userName another user already has."""
    connection = await self.connect()
    try:
        yield connection
        await connection.commit()
    except sqlite3.IntegrityError:
        await connection.rollback()
        raise UniquenessException from None

Update a resource#

update() receives the whole new state of a resource, for a PUT request and for a PATCH request. The server has already applied the request. Keep meta.created, and change meta.lastModified and meta.version.

When expected_version is given, write the resource only if its stored version is still expected_version, in the same query. Another client may change the resource between two queries. When no row changes, raise NotFoundException if the resource is gone, and PreconditionFailedException otherwise. The version is an ETag such as W/"3": the example compares its value, 3, with the version column.

def update(
    self,
    resource_type: ResourceType,
    resource: Resource[Any],
    *,
    expected_version: str | None = None,
) -> Resource[Any]:
    query = (
        "UPDATE resources SET user_name = ?, attributes = ?, last_modified = ?,"
        " version = version + 1 WHERE resource_type = ? AND id = ?"
    )
    parameters = [
        getattr(resource, "user_name", None),
        resource.model_dump_json(),
        datetime.now(UTC).isoformat(),
        resource_type.name,
        resource.id,
    ]
    if expected_version is not None:
        query += " AND version = ?"
        parameters.append(expected_version.removeprefix("W/").strip('"'))
    with self.writing():
        cursor = self.connection.execute(query, parameters)
    if cursor.rowcount == 0:
        self.get(resource_type, str(resource.id))
        raise PreconditionFailedException
    return self.get(resource_type, str(resource.id))

async def update(
    self,
    resource_type: ResourceType,
    resource: Resource[Any],
    *,
    expected_version: str | None = None,
) -> Resource[Any]:
    query = (
        "UPDATE resources SET user_name = ?, attributes = ?, last_modified = ?,"
        " version = version + 1 WHERE resource_type = ? AND id = ?"
    )
    parameters = [
        getattr(resource, "user_name", None),
        resource.model_dump_json(),
        datetime.now(UTC).isoformat(),
        resource_type.name,
        resource.id,
    ]
    if expected_version is not None:
        query += " AND version = ?"
        parameters.append(expected_version.removeprefix("W/").strip('"'))
    async with self.writing() as connection:
        cursor = await connection.execute(query, parameters)
    if cursor.rowcount == 0:
        await self.get(resource_type, str(resource.id))
        raise PreconditionFailedException
    return await self.get(resource_type, str(resource.id))

Delete a resource#

delete() removes the resource, with the same condition on expected_version in its query:

def delete(
    self,
    resource_type: ResourceType,
    resource_id: str,
    *,
    expected_version: str | None = None,
) -> None:
    query = "DELETE FROM resources WHERE resource_type = ? AND id = ?"
    parameters = [resource_type.name, resource_id]
    if expected_version is not None:
        query += " AND version = ?"
        parameters.append(expected_version.removeprefix("W/").strip('"'))
    with self.writing():
        cursor = self.connection.execute(query, parameters)
    if cursor.rowcount == 0:
        self.get(resource_type, resource_id)
        raise PreconditionFailedException

async def delete(
    self,
    resource_type: ResourceType,
    resource_id: str,
    *,
    expected_version: str | None = None,
) -> None:
    query = "DELETE FROM resources WHERE resource_type = ? AND id = ?"
    parameters = [resource_type.name, resource_id]
    if expected_version is not None:
        query += " AND version = ?"
        parameters.append(expected_version.removeprefix("W/").strip('"'))
    async with self.writing() as connection:
        cursor = await connection.execute(query, parameters)
    if cursor.rowcount == 0:
        await self.get(resource_type, resource_id)
        raise PreconditionFailedException

Search the resources#

search() receives the resource types to search, and a validated SearchRequest. A search on an endpoint covers one resource type, and a search at the root covers all of them. Filter, sort and page the resources, and return a SearchPage with the page and the number of matching resources:

def search(
    self,
    resource_types: list[ResourceType],
    search_request: SearchRequest[Any],
    *,
    position: Any = None,
) -> SearchPage:
    found = []
    for resource_type in resource_types:
        rows = self.connection.execute(
            "SELECT * FROM resources WHERE resource_type = ? ORDER BY created",
            (resource_type.name,),
        )
        found += [self.row_to_resource(resource_type, row) for row in rows]
    if search_request.filter is not None:
        found = [
            resource for resource in found if search_request.filter.match(resource)
        ]
    found = search_request.sort(found)
    page = found[search_request.start_index_0 : search_request.stop_index_0]
    return SearchPage(len(found), page)

async def search(
    self,
    resource_types: list[ResourceType],
    search_request: SearchRequest[Any],
    *,
    position: Any = None,
) -> SearchPage:
    connection = await self.connect()
    found = []
    for resource_type in resource_types:
        cursor = await connection.execute(
            "SELECT * FROM resources WHERE resource_type = ? ORDER BY created",
            (resource_type.name,),
        )
        found += [
            self.row_to_resource(resource_type, row)
            for row in await cursor.fetchall()
        ]
    if search_request.filter is not None:
        found = [
            resource for resource in found if search_request.filter.match(resource)
        ]
    found = search_request.sort(found)
    page = found[search_request.start_index_0 : search_request.stop_index_0]
    return SearchPage(len(found), page)

Count the matching resources before paging them: totalResults counts every matching resource, and the page holds at most count of them.

A search receives a position too. It locates a page when a cursor pages the resources (RFC 9865). The storage gives the positions of the next and the previous pages in the SearchPage, and the server passes them back for the next request. The cursors must give stable pages, even when resources are created or deleted between two pages: Paging the results explains why. A storage that gives them sets supports_cursors, and Page the results with cursors explains how. The storage of the example pages by index only.

Search in a database#

The search of the example reads every resource. A database storage translates the filter of the SearchRequest into a query instead:

The server already refuses a filter or a sort that its ServiceProviderConfig does not announce. A provider without config announces PATCH, and no other capability. Announce the filter, the sort and the ETags in the config of the provider once the storage supports them. A storage that cannot search several resource types at once raises NotImplementedException when it receives more than one.

Compare the strings#

The filters, the sorting and the uniqueness checks compare the strings in the form AttributeBinding.comparable returns. The ScimPolicy of the provider decides this form with its comparison_key. By default, strings are normalized to NFC, and the case of the attributes that are not caseExact is ignored. Pass another policy to the provider to compare otherwise. For instance, the precis_key function of Compare values applies the PRECIS rules:

provider = ScimProvider(
    models=...,
    resource_types=...,
    policy=ScimPolicy(comparison_key=precis_key),
)

While the server handles a request, it puts the policy of the provider in the context of the request. comparable() reads the policy from this context. The methods of the storage therefore compare as the policy says, without receiving it. The in-memory storage compares this way in its filters, its sorting and its uniqueness checks.

A database compares strings with its own rules. A storage that compares the strings in SQL gets other results, unless it stores the form comparable() returns. Document the differences for the users of the storage.

An asynchronous storage that runs blocking code with loop.run_in_executor runs it outside this context. comparable() then uses the default policy. Use asyncio.to_thread(), which keeps the context, or pass the policy to comparable().

Enclose each operation in a transaction#

The server calls operation() around each SCIM operation, and around each operation of a bulk request. Return a context manager from it to open a transaction or a savepoint. The default context does nothing. AsyncScimStorage.operation returns an asynchronous context manager.

Serve the storage#

Pass the storage to WSGIApplication or to ASGIApplication, or to the handler of Integrate a web framework:

storage = SQLiteStorage(sqlite3.connect("scim.sqlite"), provider)
app = WSGIApplication(storage, provider)

async_storage = AsyncSQLiteStorage("scim.sqlite", provider)
app = ASGIApplication(async_storage, provider)

Close the database when the application stops:

storage.connection.close()

await async_storage.close()

Writing resources explains why the storage receives the whole resource, and how the versions protect concurrent writes.