Serve an existing data model

Use this guide to serve over SCIM the data an application already keeps in its own tables. It assumes Write a storage, which explains the rules each storage method follows. The examples serve the members and the books of a library.

Describe the users

Start from the tables of the application, unchanged. The example starts from this table of members:

MEMBERS = """
CREATE TABLE IF NOT EXISTS members (
    id INTEGER PRIMARY KEY,
    login TEXT NOT NULL UNIQUE COLLATE NOCASE,
    email TEXT,
    first_name TEXT,
    last_name TEXT,
    card_number TEXT NOT NULL,
    active INTEGER NOT NULL,
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL
)
"""

Map each column of the table to a SCIM attribute. In the example, most columns have a standard attribute of User:

Column	SCIM attribute
id	id
login	userName
email	emails, as the primary email
first_name, last_name	name.givenName, name.familyName
active	active
created_at, updated_at	meta.created, meta.lastModified

Declare the columns without a standard attribute in an Extension. Mark the values that the application assigns itself as read-only: a client cannot set them. In the example, the library assigns the card numbers:

class LibraryMember(Extension):
    __schema__ = URN("urn:example:params:scim:schemas:extension:library:2.0:User")
    card_number: Annotated[str | None, Mutability.read_only] = None

Serve a custom resource type

Give the data that fits no standard resource a resource type of its own. In the example, the library also keeps its books:

BOOKS = """
CREATE TABLE IF NOT EXISTS books (
    id INTEGER PRIMARY KEY,
    title TEXT NOT NULL,
    isbn TEXT UNIQUE,
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL
)
"""

Subclass Resource, with a schema URN of the application. Declare the attributes with their SCIM characteristics, such as Required or Uniqueness:

class Book(Resource):
    __schema__ = URN("urn:example:params:scim:schemas:library:2.0:Book")
    title: Annotated[str | None, Required.true] = None
    isbn: Annotated[str | None, Uniqueness.server] = None

Declare the service

Declare every model and every resource type in the provider. The server then serves the users under /Users and the books under /Books, and describes them in /Schemas and /ResourceTypes:

>>> from myapp.scim import Book
>>> from myapp.scim import LibraryMember
>>> from scim2_models import ResourceType
>>> from scim2_models import ScimProvider
>>> from scim2_models import User
>>> from scim2_server.utils import load_default_service_provider_config

>>> provider = ScimProvider(
...     models=[User, LibraryMember, Book],
...     resource_types=[
...         ResourceType.from_resource(User[LibraryMember]),
...         ResourceType.from_resource(Book),
...     ],
...     config=load_default_service_provider_config(),
... )

Turn a row into a resource

Build the resource from the columns of a row. Use the identifier of the row as the id of the resource. Build the version from a column that changes on every write, such as updated_at in the example:

def meta(resource_type: str, row: sqlite3.Row) -> Meta:
    return Meta(
        resource_type=resource_type,
        created=row["created_at"],
        last_modified=row["updated_at"],
        version=f'W/"{row["updated_at"]}"',
    )


def row_to_user(row: sqlite3.Row) -> User[LibraryMember]:
    user = Member(
        id=str(row["id"]),
        user_name=row["login"],
        name=Name(given_name=row["first_name"], family_name=row["last_name"]),
        emails=[Email(value=row["email"], primary=True)] if row["email"] else None,
        active=bool(row["active"]),
        meta=meta("User", row),
    )
    user[LibraryMember] = LibraryMember(card_number=row["card_number"])
    return user

Each model has its own function. The books only have a title and an ISBN:

def row_to_book(row: sqlite3.Row) -> Book:
    return Book(
        id=str(row["id"]),
        title=row["title"],
        isbn=row["isbn"],
        meta=meta("Book", row),
    )

Turn a resource into a row

Return the columns that the resource sets. Leave out the attributes without a column, such as phoneNumbers: the next read returns the resource without them. Leave out the read-only attributes too, and assign them in a separate function:

def user_to_row(user: User[Any]) -> dict[str, Any]:
    """Return the columns that a user sets."""
    emails = user.emails or []
    email = next((e for e in emails if e.primary), emails[0] if emails else None)
    name = user.name or Name()
    return {
        "login": user.user_name,
        "email": email.value if email else None,
        "first_name": name.given_name,
        "last_name": name.family_name,
        "active": user.active is not False,
    }


def new_card_number() -> dict[str, Any]:
    return {"card_number": f"{secrets.randbelow(10**8):08}"}

def book_to_row(book: Book) -> dict[str, Any]:
    """Return the columns that a book sets."""
    return {"title": book.title, "isbn": book.isbn}

Choose the table of a resource type

Every storage method receives the resource type of the request. Describe the table of each resource type: its name, its two conversions, and the function that returns the values the application assigns on a creation, such as the card numbers of the members. Choose the table from the name of the resource type:

@dataclass
class Table:
    """The table of a resource type, and the conversions between its rows and the resources."""

    name: str
    to_resource: Callable[[sqlite3.Row], Resource[Any]]
    to_row: Callable[[Any], dict[str, Any]]
    assigned: Callable[[], dict[str, Any]] = dict


TABLES = {
    "User": Table("members", row_to_user, user_to_row, new_card_number),
    "Book": Table("books", row_to_book, book_to_row),
}

Write the storage

On a creation, choose the table, and build the row from the resource and from the assigned values. Let the table choose the identifier. The column names come from the code, never from the client, so the query can hold them:

Sync

def create(
    self, resource_type: ResourceType, resource: Resource[Any]
) -> Resource[Any]:
    table = TABLES[resource_type.name]
    now = datetime.now(UTC).isoformat()
    row = {
        **table.to_row(resource),
        **table.assigned(),
        "created_at": now,
        "updated_at": now,
    }
    columns = ", ".join(row)
    values = ", ".join(f":{column}" for column in row)
    with self.writing():
        cursor = self.connection.execute(
            f"INSERT INTO {table.name} ({columns}) VALUES ({values})", row
        )
    return self.get(resource_type, str(cursor.lastrowid))

Async

async def create(
    self, resource_type: ResourceType, resource: Resource[Any]
) -> Resource[Any]:
    table = TABLES[resource_type.name]
    now = datetime.now(UTC).isoformat()
    row = {
        **table.to_row(resource),
        **table.assigned(),
        "created_at": now,
        "updated_at": now,
    }
    columns = ", ".join(row)
    values = ", ".join(f":{column}" for column in row)
    async with self.writing() as connection:
        cursor = await connection.execute(
            f"INSERT INTO {table.name} ({columns}) VALUES ({values})", row
        )
    return await self.get(resource_type, str(cursor.lastrowid))

On an update, write the columns of the new state. The assigned values keep their stored value. The query checks expected_version against the updated_at column, as Write a storage does with its version column:

Sync

def update(
    self,
    resource_type: ResourceType,
    resource: Resource[Any],
    *,
    expected_version: str | None = None,
) -> Resource[Any]:
    table = TABLES[resource_type.name]
    row = {**table.to_row(resource), "updated_at": datetime.now(UTC).isoformat()}
    assignments = ", ".join(f"{column} = :{column}" for column in row)
    query = f"UPDATE {table.name} SET {assignments} WHERE id = :id"
    parameters = {**row, "id": resource.id}
    if expected_version is not None:
        query += " AND updated_at = :version"
        parameters["version"] = 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

async def update(
    self,
    resource_type: ResourceType,
    resource: Resource[Any],
    *,
    expected_version: str | None = None,
) -> Resource[Any]:
    table = TABLES[resource_type.name]
    row = {**table.to_row(resource), "updated_at": datetime.now(UTC).isoformat()}
    assignments = ", ".join(f"{column} = :{column}" for column in row)
    query = f"UPDATE {table.name} SET {assignments} WHERE id = :id"
    parameters = {**row, "id": resource.id}
    if expected_version is not None:
        query += " AND updated_at = :version"
        parameters["version"] = 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))

Write the other methods as in Write a storage, with the table of the resource type: get reads a row, delete removes it, and search reads the rows of each resource type it receives.

Serve the storage

Pass the storage and the provider to the application, with the database of the application:

Sync

connection = sqlite3.connect("library.sqlite")
app = WSGIApplication(LibraryStorage(connection), provider)

Async

app = ASGIApplication(AsyncLibraryStorage("library.sqlite"), provider)

The rows already in the tables are served at once: the member of row 1 is the user /v2/Users/1.

Integrate a web framework serves the storage from the web framework of the application.