crossauth_backend.storageimpl package

Submodules

crossauth_backend.storageimpl.inmemorystorage module

class crossauth_backend.storageimpl.inmemorystorage.InMemoryKeyStorage

Bases: KeyStorage

Implementation of KeyStorage where keys stored in memory. Intended for testing.

async delete_all_for_user(userid: str | int | None, prefix: str, except_key: str | None = None) → None

Deletes all keys from storage for the given user ID

Parameters:

• userid (int|str|None) – User ID to delete keys for

• prefix (str) – Only keys starting with this prefix will be deleted

• except_key (str|None) – If defined, the key with this value will not be deleted

async delete_data(key_name: str, data_name: str) → None

The ‘data’ field in a key entry is a JSON string. This method should atomically delete a field in it.

Parameters:

• key_name (str) – The name of the key to update, as it appears in the table.

• data_name (str) – The field name to delete. This can contain dots, e.g., ‘part1.part2’, which means ‘part2’ within ‘part1’ is deleted.

async delete_key(value: str) → None

Deletes a key from storage (e.g., the database).

Parameters:

value (str) – The key to delete

async delete_matching(key: PartialKey) → None

Deletes all matching the given specs

Parameters:

key (crossauth_backend.PartialKey) – Any key matching all defined values in this object will be deleted

async get_all_for_user(userid: str | int | None = None) → List[Key]

Return all keys matching the given user ID

Parameters:

userid (str|int|None) – User to return keys for

Returns:

List[Key]: An array of keys

async get_key(key: str) → Key

Returns the matching key in the session storage or raises an exception if it doesn’t exist.

Parameters:

key (ster) – The key to look up, as it will appear in this storage (typically unsigned, hashed)

Returns:

The matching Key record.

print()

async save_key(userid: str | int | None, value: str, date_created: datetime, expires: datetime | None = None, data: str | None = None, extra_fields: Mapping[str, Any] | None = None) → None

Saves a session key in the session storage (e.g., database).

Parameters:

• userid (int|str|None) – The ID of the user. This matches the primary key in the UserStorage implementation.

• value (str) – The key value to store.

• date_created (datetime) – The date/time the key was created.

• expires (datetime|None) – The date/time the key expires.

• extra_fields (Mapping[str, Any]|None) – These will also be saved in the key record

Padam str|None data:

An optional value, specific to the type of key, e.g., new email for email change tokens

async update_data(key_name: str, data_name: str, value: Any | None) → None

The ‘data’ field in a key entry is a JSON string. This method should atomically update a field in it.

Parameters:

• key_name (str) – The name of the key to update, as it appears in the table.

• data_name (str) – The field name to update. This can contain dots, e.g., ‘part1.part2’, which means ‘part2’ within ‘part1’ is updated.

• value (Any|None) – The new value.

async update_key(key: PartialKey) → None

If the given session key exists in the database, update it with the passed values. If it doesn’t exist, raise a CrossauthError with ErrorCode ‘InvalidKey’.

Parameters:

key (crossauth_backend.PartialKey) – The fields defined in this will be updated. ‘id’ must be present and it will not be updated.

async update_many_data(key_name: str, data_array: List[KeyDataEntry]) → None

Same as ‘update_data’ but updates several keys.

Ensure it is done as a single transaction.

Parameters:

• key_name (str) – The key to update

• data_array (List[crossauth_backend.KeyDataEntry) – dataName and value pairs

class crossauth_backend.storageimpl.inmemorystorage.InMemoryOAuthAuthorizationStorage(options: OAuthAuthorizationStorageOptions = {})

Bases: OAuthAuthorizationStorage

Implementation of KeyStorage where keys stored in memory. Intended for testing.

async get_authorizations(client_id: str, userid: str | int | None = None) → List[str | None]

Returns the matching all scopes authorized for the given client and optionally user.

Parameters:

• client_id (str) – the client_id to look up

• userid (int|str|None) – the userid to look up, None for a client authorization not user authorization

Returns:

The authorized scopes as a list.

async update_authorizations(client_id: str, userid: str | int | None, authorizations: List[str | None]) → None

Saves a new set of authorizations for the given client and optionally user.

Deletes the old ones.

Parameters:

• client_id (str) – the client_id to look up

• userid (str|int|None) – the userid to look up, None for a client authorization not user authorization

• List[str|None]authorizations – new set of authorized scopes, which may be empty

class crossauth_backend.storageimpl.inmemorystorage.InMemoryOAuthClientStorage(options: OAuthClientStorageOptions = {})

Bases: OAuthClientStorage

Implementation of KeyStorage where keys stored in memory. Intended for testing.

async create_client(client: OAuthClient) → OAuthClient

Creates and returns a new client with random ID and optionally secret.

Saves in the database.

Parameters:

client (crossauth_backend.OAuthClient) – the client to save.

Returns:

The new client.

async delete_client(client_id: str) → None

Deletes a key from storage.

Parameters:

client_id (str) – the client to delete

async get_client_by_id(client_id: str) → OAuthClient

Returns the matching client by its auto-generated id in the storage or throws an exception if it doesn’t exist.

Parameters:

client_id (str) – the client_id to look up

Returns:

he matching OAuthClient object.

async get_client_by_name(name: str, userid: str | int | None | NullType = None) → List[OAuthClient]

Returns the matching client in the storage by friendly name or throws an exception if it doesn’t exist.

Parameters:

• name (str) – the client name to look up

• userid (str|int|None) – if defined, only return clients belonging to this user. if None, return only clients with a null userid. if not provided, return all clients with this name.

Returns:

A list of OAuthClient objects.

:raises crossauth_backend.CrossauthError: with ErrorCode of ‘InvalidSessionId’ if a match was not found in session storage.

async get_clients(skip: int | None = None, take: int | None = None, userid: str | int | None | NullType = None) → List[OAuthClient]

Returns all clients in alphabetical order of client name.

Parameters:

• skip (int|None) – skip this number of records from the start in alphabetical order

• take (int|None) – return at most this number of records

• userid (str|int|None) – if defined, only return clients belonging to this user. if None, return only clients with a null userid. if not provided, return all clients.

Returns:

A list of OAuthClient objects.

async update_client(client: PartialOAuthClient) → None

If the given session key exists in the database, update it with the passed values. If it doesn’t exist, throw a CrossauthError with ‘InvalidClient’.

Parameters:

client (crossauth_backend.OAuthClient) – all fields to update (client_id must be set but will not be updated)

:raises crossauth_backend.CrossauthError: with ‘InvalidClient’ if the client doesn’t exist.

class crossauth_backend.storageimpl.inmemorystorage.InMemoryUserStorage(options: InMemoryUserStorageOptions = {})

Bases: UserStorage

Implementation of KeyStorage where keys stored in memory. Intended for testing.

async create_user(user: UserInputFields, secrets: UserSecretsInputFields | None = None) → User

Creates a user with the given details and secrets.

Parameters:

• user (crossauth_backend.UserInputFields) – will be put in the User table

• secrets (crossauth_backend.UserSecretsInputFields|None) – will be put in the UserSecrets table

Returns:

the new user as a User object

:raises crossauth_backend.CrossauthError with ErrorCode Configuration

async delete_user_by_id(id: str | int) → None

If the storage supports this, delete the user with the given ID from storage.

Parameters:

id (str|int) – id of user to delete

async delete_user_by_username(username: str) → None

If the storage supports this, delete the named user from storage.

Parameters:

username (str) – username to delete

async get_user_by(field: str, value: str, options: UserStorageGetOptions = {}) → UserAndSecrets

Returns user matching the given field, or throws an exception.

This does no normalisation. Currently it is only used for the OAuth client if you set user_creation_type to “merge”, “embed” or “custom”.

Parameters:

• field (str) – the field to match

• value (str) – the value to match

• options (UserStorageGetOptions) – optionally turn off checks. Used internally

:raises crossauth_backend.CrossauthError: with ErrorCode either UserNotExist or Connection

async get_user_by_email(email: str, options: UserStorageGetOptions = {}) → UserAndSecrets

Returns user matching the given email address, or throws an exception.

If normalize_email is true, email should be matched normalized and lowercased (using normalize()) If the email field doesn’t exist, username is assumed to be the email column

Parameters:

• email – the email address to return the user of

• options – optionally turn off checks. Used internally

Raises:

crossauth_backend.CrossauthError – with ErrorCode either UserNotExist or Connection

async get_user_by_id(id: str | int, options: UserStorageGetOptions = {}) → UserAndSecrets

Returns user matching the given user id, or throws an exception.

Note that implementations are free to define what the user ID is. It can be a number or string, or can simply be username.

Parameters:

• id (str|int) – the user id to return the user of

• options (UserStorageGetOptions) – optionally turn off checks. Used internally

:raises crossauth_backend.CrossauthError with ErrorCode either UserNotExist or Connection

async get_user_by_username(username: str, options: UserStorageGetOptions = {}) → UserAndSecrets

Returns user matching the given username, or throws an exception.

If normalize_username is true, the username should be matched normalized and lowercased (using normalize())

Parameters:

• username (str) – the username to return the user of

• options (UserStorageGetOptions) – optionally turn off checks. Used internally

:raises crossauth_backend.CrossauthError: with ErrorCode either UserNotExist or Connection

async get_users(skip: int | None = None, take: int | None = None) → List[User]

Returns all users in the storage, in a fixed order defined by the storage (e.g. alphabetical by username)

Parameters:

• skip (int|None) – skip this number of records from the start of the set

• take (int|None) – only return at most this number of records

Returns:

an array of User objects

async update_user(user: PartialUser, secrets: PartialUserSecrets | None = None) → None

Updates an existing user with the given details and secrets.

If the given user exists in the database, update it with the passed values. If it doesn’t exist, throw a CrossauthError with ErrorCode InvalidKey.

Parameters:

• user (crossauth_backend.PartialUser) – The ‘id’ field must be set, but all others are optional. Any parameter not set (or None) will not be updated. If you want to set something to None in the database, pass the value as None, not undefined.

• secrets (crossauth_backend.PartialUserSecrets|None) – Optional secrets to update

class crossauth_backend.storageimpl.inmemorystorage.InMemoryUserStorageOptions

Bases: UserStorageOptions

admin_editable_fields: List[str]

normalize_email: bool

normalize_username: bool

user_editable_fields: List[str]

crossauth_backend.storageimpl.ldapstorage module

class crossauth_backend.storageimpl.ldapstorage.LdapUserStorage(local_storage: UserStorage, options: LdapUserStorageOptions = {})

Bases: UserStorage

Wraps another user storage but with the authentication done in LDAP.

This class still needs a user to be created in another database, with for example a user id that can be referenced in key storage, and a state variable.

An admin account is not used. Searches are done as the user, with the user’s password.

async create_user(user: UserInputFields, secrets: UserSecretsInputFields | None = None) → User

Creates a user with the given details and secrets.

Parameters:

• user (crossauth_backend.UserInputFields) – will be put in the User table

• secrets (crossauth_backend.UserSecretsInputFields|None) – will be put in the UserSecrets table

Returns:

the new user as a User object

:raises crossauth_backend.CrossauthError with ErrorCode Configuration

async delete_user_by_id(id: str | int) → None

If the storage supports this, delete the user with the given ID from storage.

Parameters:

id (str|int) – id of user to delete

async delete_user_by_username(username: str) → None

If the storage supports this, delete the named user from storage.

Parameters:

username (str) – username to delete

async get_ldap_user(username: str, password: str) → LdapUser

Gets the user from LDAP. Does not check local storage.

If the user doesn’t exist or authentication fails, an exception is thrown :param username: the username to fetch :param password: the LDAP password :returns: the matching LdapUser :raises: CrossauthError with ErrorCode UsernameOrPasswordInvalid or Connection

async get_user_by(field: str, value: str, options: UserStorageGetOptions = {}) → UserAndSecrets

Returns user matching the given field, or throws an exception.

This does no normalisation. Currently it is only used for the OAuth client if you set user_creation_type to “merge”, “embed” or “custom”.

Parameters:

• field (str) – the field to match

• value (str) – the value to match

• options (UserStorageGetOptions) – optionally turn off checks. Used internally

:raises crossauth_backend.CrossauthError: with ErrorCode either UserNotExist or Connection

async get_user_by_email(email: str, options: UserStorageGetOptions = {}) → UserAndSecrets

Returns user matching the given email address, or throws an exception.

If normalize_email is true, email should be matched normalized and lowercased (using normalize()) If the email field doesn’t exist, username is assumed to be the email column

Parameters:

• email – the email address to return the user of

• options – optionally turn off checks. Used internally

Raises:

crossauth_backend.CrossauthError – with ErrorCode either UserNotExist or Connection

async get_user_by_id(id: str | int, options: UserStorageGetOptions = {}) → UserAndSecrets

Returns user matching the given user id, or throws an exception.

Note that implementations are free to define what the user ID is. It can be a number or string, or can simply be username.

Parameters:

• id (str|int) – the user id to return the user of

• options (UserStorageGetOptions) – optionally turn off checks. Used internally

:raises crossauth_backend.CrossauthError with ErrorCode either UserNotExist or Connection

async get_user_by_username(username: str, options: UserStorageGetOptions = {}) → UserAndSecrets

Returns user matching the given username, or throws an exception.

If normalize_username is true, the username should be matched normalized and lowercased (using normalize())

Parameters:

• username (str) – the username to return the user of

• options (UserStorageGetOptions) – optionally turn off checks. Used internally

:raises crossauth_backend.CrossauthError: with ErrorCode either UserNotExist or Connection

async get_users(skip: int | None = None, take: int | None = None) → List[User]

Returns all users in the storage, in a fixed order defined by the storage (e.g. alphabetical by username)

Parameters:

• skip (int|None) – skip this number of records from the start of the set

• take (int|None) – only return at most this number of records

Returns:

an array of User objects

async ldap_bind(dn: str, password: str) → Connection

bind and return the ldap client from https://github.com/shaozi/ldap-authentication/blob/master/index.js

require_user_entry() → bool

Returns false

async search_user(ldap_client: Connection, user_dn: str, attributes: List[str] | None = None) → LdapUser

Search for user in LDAP

async update_user(user: PartialUser, secrets: PartialUserSecrets | None = None) → None

Updates an existing user with the given details and secrets.

If the given user exists in the database, update it with the passed values. If it doesn’t exist, throw a CrossauthError with ErrorCode InvalidKey.

Parameters:

• user (crossauth_backend.PartialUser) – The ‘id’ field must be set, but all others are optional. Any parameter not set (or None) will not be updated. If you want to set something to None in the database, pass the value as None, not undefined.

• secrets (crossauth_backend.PartialUserSecrets|None) – Optional secrets to update

class crossauth_backend.storageimpl.ldapstorage.LdapUserStorageOptions

Bases: UserStorageOptions

Optional parameters for {@link LdapUserStorage}.

admin_editable_fields: List[str]

create_user_fn: Callable[[UserInputFields, LdapUser], UserInputFields]

A function to create a user object given the entry in LDAP and additional fields. The additional fields might be useful for attributes that aren’t in LDAP and the user needs to be prompted for, for example email address. The default function sets username to uid from ldapUser, state to active and takes every field for user (overriding status and username if present).

ldap_urls: List[str]

//ldap,example.com:1636 No default (required)

Type:

Utl running LDAP server. eg ldap

Type:

//ldap.example.com or ldaps

ldap_user_search_base: str

Search base, for user queries, eg ou=users,dc=example,dc=com. Default empty

ldap_username_attribute: str

Username attribute for searches. Default “cn”.

normalize_email: bool

normalize_username: bool

user_editable_fields: List[str]

class crossauth_backend.storageimpl.ldapstorage.LdapUserWithState

Bases: LdapUser

dn: str

state: str

uid: NotRequired[str | List[str]]

crossauth_backend.storageimpl.ldapstorage.default_create_user_dn(user: UserInputFields, ldap_user: LdapUser) → UserInputFields

crossauth_backend.storageimpl.sqlalchemystorage module

class crossauth_backend.storageimpl.sqlalchemystorage.SqlAlchemyKeyStorage(engine: AsyncEngine, options: SqlAlchemyKeyStorageOptions = {})

Bases: KeyStorage

async delete_all_for_user(userid: str | int | None, prefix: str, except_key: str | None = None) → None

Deletes all keys from storage for the given user ID

Parameters:

• userid (int|str|None) – User ID to delete keys for

• prefix (str) – Only keys starting with this prefix will be deleted

• except_key (str|None) – If defined, the key with this value will not be deleted

async delete_data(key_name: str, data_name: str) → None

The ‘data’ field in a key entry is a JSON string. This method should atomically delete a field in it.

Parameters:

• key_name (str) – The name of the key to update, as it appears in the table.

• data_name (str) – The field name to delete. This can contain dots, e.g., ‘part1.part2’, which means ‘part2’ within ‘part1’ is deleted.

async delete_key(value: str) → None

Deletes a key from storage (e.g., the database).

Parameters:

value (str) – The key to delete

async delete_matching(key: PartialKey) → None

Deletes all matching the given specs

Parameters:

key (crossauth_backend.PartialKey) – Any key matching all defined values in this object will be deleted

async delete_with_prefix(userid: str | int | None, prefix: str) → None

async get_all_for_user(userid: str | int | None = None) → List[Key]

Return all keys matching the given user ID

Parameters:

userid (str|int|None) – User to return keys for

Returns:

List[Key]: An array of keys

async get_key(key: str) → Key

Returns the matching key in the session storage or raises an exception if it doesn’t exist.

Parameters:

key (ster) – The key to look up, as it will appear in this storage (typically unsigned, hashed)

Returns:

The matching Key record.

async get_key_in_transaction(conn: AsyncConnection, keyValue: str) → Key

async save_key(userid: str | int | None, value: str, date_created: datetime, expires: datetime | None = None, data: str | None = None, extra_fields: Mapping[str, Any] | None = None) → None

Saves a session key in the session storage (e.g., database).

Parameters:

• userid (int|str|None) – The ID of the user. This matches the primary key in the UserStorage implementation.

• value (str) – The key value to store.

• date_created (datetime) – The date/time the key was created.

• expires (datetime|None) – The date/time the key expires.

• extra_fields (Mapping[str, Any]|None) – These will also be saved in the key record

Padam str|None data:

An optional value, specific to the type of key, e.g., new email for email change tokens

to_dict(row: Row[Any], with_relationships: bool = True) → dict[str, Any]

async update_data(key_name: str, data_name: str, value: Any) → None

The ‘data’ field in a key entry is a JSON string. This method should atomically update a field in it.

Parameters:

• key_name (str) – The name of the key to update, as it appears in the table.

• data_name (str) – The field name to update. This can contain dots, e.g., ‘part1.part2’, which means ‘part2’ within ‘part1’ is updated.

• value (Any|None) – The new value.

async update_key(key: PartialKey) → None

If the given session key exists in the database, update it with the passed values. If it doesn’t exist, raise a CrossauthError with ErrorCode ‘InvalidKey’.

Parameters:

key (crossauth_backend.PartialKey) – The fields defined in this will be updated. ‘id’ must be present and it will not be updated.

async update_key_in_transaction(conn: AsyncConnection, key: PartialKey) → None

async update_many_data(key_name: str, data_array: List[KeyDataEntry]) → None

Same as ‘update_data’ but updates several keys.

Ensure it is done as a single transaction.

Parameters:

• key_name (str) – The key to update

• data_array (List[crossauth_backend.KeyDataEntry) – dataName and value pairs

class crossauth_backend.storageimpl.sqlalchemystorage.SqlAlchemyKeyStorageOptions

Bases: TypedDict

Optional parameters for :class: SqlAlchemyKeyStorage.

See :func: SqlAlchemyKeyStorage__init__ for details

key_table: str

userid_foreign_key_column: str

class crossauth_backend.storageimpl.sqlalchemystorage.SqlAlchemyOAuthAuthorizationStorage(engine: AsyncEngine, options: SqlAlchemyOAuthAuthorizationStorageOptions = {})

Bases: OAuthAuthorizationStorage

async get_authorizations(client_id: str, userid: str | int | None = None) → List[str | None]

Returns the matching all scopes authorized for the given client and optionally user.

Parameters:

• client_id (str) – the client_id to look up

• userid (int|str|None) – the userid to look up, None for a client authorization not user authorization

Returns:

The authorized scopes as a list.

async update_authorizations(client_id: str, userid: str | int | None, authorizations: List[str | None]) → None

Saves a new set of authorizations for the given client and optionally user.

Deletes the old ones.

Parameters:

• client_id (str) – the client_id to look up

• userid (str|int|None) – the userid to look up, None for a client authorization not user authorization

• List[str|None]authorizations – new set of authorized scopes, which may be empty

class crossauth_backend.storageimpl.sqlalchemystorage.SqlAlchemyOAuthAuthorizationStorageOptions

Bases: OAuthAuthorizationStorageOptions

Optional parameters for :class: SqlAlchemyUserStorage.

See :func: SqlAlchemyUserStorage__init__ for details

azuthorization_table: str

Name of client table. Default OAuthClient

userid_foreign_key_column: str

Name of the user id column in the table. Default userid

class crossauth_backend.storageimpl.sqlalchemystorage.SqlAlchemyOAuthClientStorage(engine: AsyncEngine, options: SqlAlchemyOAuthClientStorageOptions = {})

Bases: OAuthClientStorage

async create_client(client: OAuthClient) → OAuthClient

Creates and returns a new client with random ID and optionally secret.

Saves in the database.

Parameters:

client (crossauth_backend.OAuthClient) – the client to save.

Returns:

The new client.

async delete_client(client_id: str) → None

Deletes a key from storage.

Parameters:

client_id (str) – the client to delete

async get_client_by_id(client_id: str) → OAuthClient

Returns the matching client by its auto-generated id in the storage or throws an exception if it doesn’t exist.

Parameters:

client_id (str) – the client_id to look up

Returns:

he matching OAuthClient object.

async get_client_by_name(name: str, userid: str | int | None | NullType = None) → List[OAuthClient]

Returns the matching client in the storage by friendly name or throws an exception if it doesn’t exist.

Parameters:

• name (str) – the client name to look up

• userid (str|int|None) – if defined, only return clients belonging to this user. if None, return only clients with a null userid. if not provided, return all clients with this name.

Returns:

A list of OAuthClient objects.

:raises crossauth_backend.CrossauthError: with ErrorCode of ‘InvalidSessionId’ if a match was not found in session storage.

async get_client_in_transaction(conn: AsyncConnection, field: str | None, value: str | None, userid: int | str | NullType | None = None, skip: int | None = None, take: int | None = None) → List[OAuthClient]

async get_clients(skip: int | None = None, take: int | None = None, userid: str | int | None | NullType = None) → List[OAuthClient]

Returns all clients in alphabetical order of client name.

Parameters:

• skip (int|None) – skip this number of records from the start in alphabetical order

• take (int|None) – return at most this number of records

• userid (str|int|None) – if defined, only return clients belonging to this user. if None, return only clients with a null userid. if not provided, return all clients.

Returns:

A list of OAuthClient objects.

async update_client(client: PartialOAuthClient) → None

If the given session key exists in the database, update it with the passed values. If it doesn’t exist, throw a CrossauthError with ‘InvalidClient’.

Parameters:

client (crossauth_backend.OAuthClient) – all fields to update (client_id must be set but will not be updated)

:raises crossauth_backend.CrossauthError: with ‘InvalidClient’ if the client doesn’t exist.

class crossauth_backend.storageimpl.sqlalchemystorage.SqlAlchemyOAuthClientStorageOptions

Bases: OAuthClientStorageOptions

Optional parameters for :class: SqlAlchemyUserStorage.

See :func: SqlAlchemyUserStorage__init__ for details

client_table: str

Name of client table. Default OAuthClient

redirect_uri_table: str

Name of the redirect uri table. Default OAuthClientRedirectUri.

userid_foreign_key_column: str

Column name for the userid field in the client table. Default userid

valid_flow_table: str

Name of the valid flows table. Default OAuthClientValidFlow

class crossauth_backend.storageimpl.sqlalchemystorage.SqlAlchemyUserStorage(engine: AsyncEngine, options: SqlAlchemyUserStorageOptions = {})

Bases: UserStorage

async create_user(user: UserInputFields, secrets: UserSecretsInputFields | None = None) → User

Creates a user with the given details and secrets.

Parameters:

• user (crossauth_backend.UserInputFields) – will be put in the User table

• secrets (crossauth_backend.UserSecretsInputFields|None) – will be put in the UserSecrets table

Returns:

the new user as a User object

:raises crossauth_backend.CrossauthError with ErrorCode Configuration

async delete_user_by_id(id: str | int) → None

If the storage supports this, delete the user with the given ID from storage.

Parameters:

id (str|int) – id of user to delete

async delete_user_by_username(username: str) → None

If the storage supports this, delete the named user from storage.

Parameters:

username (str) – username to delete

async get_user_by(field: str, value: str | int, options: UserStorageGetOptions = {}) → UserAndSecrets

Returns user matching the given field, or throws an exception.

This does no normalisation. Currently it is only used for the OAuth client if you set user_creation_type to “merge”, “embed” or “custom”.

Parameters:

• field (str) – the field to match

• value (str) – the value to match

• options (UserStorageGetOptions) – optionally turn off checks. Used internally

:raises crossauth_backend.CrossauthError: with ErrorCode either UserNotExist or Connection

async get_user_by_email(email: str, options: UserStorageGetOptions = {}) → UserAndSecrets

Returns user matching the given email address, or throws an exception.

If normalize_email is true, email should be matched normalized and lowercased (using normalize()) If the email field doesn’t exist, username is assumed to be the email column

Parameters:

• email – the email address to return the user of

• options – optionally turn off checks. Used internally

Raises:

crossauth_backend.CrossauthError – with ErrorCode either UserNotExist or Connection

async get_user_by_id(id: str | int, options: UserStorageGetOptions = {}) → UserAndSecrets

Returns user matching the given user id, or throws an exception.

Note that implementations are free to define what the user ID is. It can be a number or string, or can simply be username.

Parameters:

• id (str|int) – the user id to return the user of

• options (UserStorageGetOptions) – optionally turn off checks. Used internally

:raises crossauth_backend.CrossauthError with ErrorCode either UserNotExist or Connection

async get_user_by_in_transaction(conn: AsyncConnection, field: str, value: str | int) → UserAndSecrets

async get_user_by_username(username: str, options: UserStorageGetOptions = {}) → UserAndSecrets

Returns user matching the given username, or throws an exception.

If normalize_username is true, the username should be matched normalized and lowercased (using normalize())

Parameters:

• username (str) – the username to return the user of

• options (UserStorageGetOptions) – optionally turn off checks. Used internally

:raises crossauth_backend.CrossauthError: with ErrorCode either UserNotExist or Connection

async get_users(skip: int | None = None, take: int | None = None) → List[User]

Returns all users in the storage, in a fixed order defined by the storage (e.g. alphabetical by username)

Parameters:

• skip (int|None) – skip this number of records from the start of the set

• take (int|None) – only return at most this number of records

Returns:

an array of User objects

to_dict(row: Row[Any]) → dict[str, Any]

async update_user(user: PartialUser, secrets: PartialUserSecrets | None = None) → None

Updates an existing user with the given details and secrets.

If the given user exists in the database, update it with the passed values. If it doesn’t exist, throw a CrossauthError with ErrorCode InvalidKey.

Parameters:

• user (crossauth_backend.PartialUser) – The ‘id’ field must be set, but all others are optional. Any parameter not set (or None) will not be updated. If you want to set something to None in the database, pass the value as None, not undefined.

• secrets (crossauth_backend.PartialUserSecrets|None) – Optional secrets to update

class crossauth_backend.storageimpl.sqlalchemystorage.SqlAlchemyUserStorageOptions

Bases: UserStorageOptions

Optional parameters for :class: SqlAlchemyUserStorage.

See :func: SqlAlchemyUserStorage__init__ for details

admin_editable_fields: List[str]

force_id_to_number: bool

This works around a Fastify and Sveltekit limitation. If the id passed to getUserById() is a string but is numeric, first try forcing it to a number before selecting. If that fails, try it as the string, Default true.

id_column: str

Name of the id column in the user table. Can be set to username if that is your primary key. Default id.

joins: List[str]

Other tables to join. Default is [] (UserSecrets is alwayws joined)

normalize_email: bool

normalize_username: bool

user_editable_fields: List[str]

user_secrets_table: str

Name of user secrets table (Default UserSecrets

user_table: str

Name of user table Default User

userid_foreign_key_column: str

Name of the user id column in the user secrets. Default userid.

crossauth_backend.storageimpl.sqlalchemystorage.adapt_date_iso_int(val: datetime)

Adapt datetime.date to ISO 8601 date.

crossauth_backend.storageimpl.sqlalchemystorage.adapt_date_iso_real(val: datetime)

Adapt datetime.date to ISO 8601 date.

crossauth_backend.storageimpl.sqlalchemystorage.adapt_datetime_epoch_int(val: datetime)

Adapt datetime.datetime to Unix timestamp.

crossauth_backend.storageimpl.sqlalchemystorage.adapt_datetime_epoch_real(val: datetime)

Adapt datetime.datetime to Unix timestamp.

crossauth_backend.storageimpl.sqlalchemystorage.adapt_datetime_iso_int(val: datetime)

Adapt datetime.datetime to timezone-naive ISO 8601 date.

crossauth_backend.storageimpl.sqlalchemystorage.adapt_datetime_iso_real(val: datetime)

Adapt datetime.datetime to timezone-naive ISO 8601 date.

crossauth_backend.storageimpl.sqlalchemystorage.convert_date_int(val: bytes | str)

Convert ISO 8601 date to datetime.date object.

crossauth_backend.storageimpl.sqlalchemystorage.convert_date_real(val: str | bytes)

Convert ISO 8601 date to datetime.date object.

crossauth_backend.storageimpl.sqlalchemystorage.convert_datetime_int(val: bytes | str)

Convert ISO 8601 datetime to datetime.datetime object.

crossauth_backend.storageimpl.sqlalchemystorage.convert_datetime_real(val: str | bytes)

Convert ISO 8601 datetime to datetime.datetime object.

crossauth_backend.storageimpl.sqlalchemystorage.convert_timestamp_int(val: int)

Convert Unix epoch timestamp to datetime.datetime object.

crossauth_backend.storageimpl.sqlalchemystorage.convert_timestamp_real(val: float)

Convert Unix epoch timestamp to datetime.datetime object.

crossauth_backend.storageimpl.sqlalchemystorage.register_sqlite_datetime()

SQLite has no date column types. As of Python 3.12, the default adapters which made this seamless don’t work.

If you don’t have your own adapters and you want to store dates etc as real, call this function to register the ones supplied with Crossauth and declare your columns as date/datetime/timestamp as normal.

Module contents