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 its case:

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.

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) -> 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) -> 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)

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. 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 the number of matching resources with the page:

def search(
    self, resource_types: list[ResourceType], search_request: SearchRequest[Any]
) -> tuple[int, list[Resource[Any]]]:
    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)
    start = (search_request.start_index or 1) - 1
    stop = None if search_request.count is None else start + search_request.count
    return len(found), found[start:stop]

async def search(
    self, resource_types: list[ResourceType], search_request: SearchRequest[Any]
) -> tuple[int, list[Resource[Any]]]:
    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)
    start = (search_request.start_index or 1) - 1
    stop = None if search_request.count is None else start + search_request.count
    return len(found), found[start:stop]

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

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 storage that cannot search several resource types at once raises NotImplementedException when it receives more than one.

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.