User Service

litestar_users.service.BaseUserService

Bases: Generic[SQLAUserT, SQLARoleT, SQLAOAuthAccountT]

Main user management interface.

__init__

__init__(secret, user_auth_identifier, user_repository, hash_schemes=None, role_repository=None, oauth2_repository=None, require_verification_on_registration=True)

User service constructor.

Parameters:

Name	Type	Description	Default
secret	str	Secret string for securely signing tokens.	required
user_auth_identifier	str	The User model attribute to identify the user during authorization.	required
user_repository	SQLAlchemyAsyncRepository[SQLAUserT]	A UserRepository instance.	required
hash_schemes	Sequence[str] | None	Schemes to use for password encryption.	None
role_repository	SQLAlchemyAsyncRepository[SQLARoleT] | None	A RoleRepository instance.	None
oauth2_repository	SQLAlchemyAsyncRepository[SQLAOAuthAccountT] | None	A OAuth2Repository instance.	None
require_verification_on_registration	bool	Whether the registration of a new user requires verification.	True

add_user async

add_user(user, verify=False, activate=True)

Create a new user programmatically.

Parameters:

Name	Type	Description	Default
user	SQLAUserT	User model instance.	required
verify	bool	Set the user's verification status to this value.	False
activate	bool	Set the user's active status to this value.	True

register async

register(data, request=None)

Register a new user and optionally run custom business logic.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	User creation data transfer object.	required
request	Request | None	The litestar request that initiated the action.	None

get_user async

get_user(user_id, load=None, execution_options=None)

Retrieve a user from the database by id.

Parameters:

Name	Type	Description	Default
user_id	UUID | int	UUID corresponding to a user primary key.	required
load	LoadSpec | None	Set relationships to be loaded	None
execution_options	dict[str, Any] | None	Set default execution options	None

get_user_by async

get_user_by(load=None, execution_options=None, **kwargs)

Retrieve a user from the database by arbitrary keyword arguments.

Parameters:

Name	Type	Description	Default
load	LoadSpec | None	Set relationships to be loaded	None
execution_options	dict[str, Any] | None	Set default execution options	None
**kwargs	Any	Keyword arguments to pass as filters.	{}

Examples:

service = UserService(...)
john = await service.get_one(email="john@example.com")

update_user async

update_user(user)

Update a user's attributes in the database.

The user instance may be partial and will be forwarded to [SQLAlchemyAsyncRepository.update][advanced_alchemy.repository.SQLAlchemyAsyncRepository.update], which ensures that only explicitly set fields are merged into the existing database record. The caller is responsible for setting user.id before invoking this method so the repository can identify which record to update.

Parameters:

Name	Type	Description	Default
user	SQLAUserT	A User model instance. Must have id set. If password_hash is set it is treated as a plain-text password and will be hashed before being stored.	required

delete_user async

delete_user(user_id)

Delete a user from the database.

Parameters:

Name	Type	Description	Default
user_id	UUID | int	UUID corresponding to a user primary key.	required

authenticate async

authenticate(data, request=None)

Authenticate a user.

Parameters:

Name	Type	Description	Default
data	Any	User authentication data transfer object.	required
request	Request | None	The litestar request that initiated the action.	None

get_additional_auth_filters

get_additional_auth_filters(data, request=None)

Get additional filters to apply to the authentication query.

By default, authentication looks up a user solely by the configured user_auth_identifier (e.g. email). In certain scenarios - such as multitenant applications - the database may legitimately contain multiple rows sharing the same identifier value, one per tenant. Because authentication has no way to narrow the lookup by user id (the id is not yet known at that point), this method exists as the extension point for supplying the extra predicates needed.

The request parameter makes it straightforward to derive filter values from the incoming HTTP context. For example, to scope authentication to the tenant identified by a custom request header::

def get_additional_auth_filters(
    self, data: Any, request: Request | None = None
) -> Sequence[ColumnElement[bool]]:
    company_id = request.headers.get("x-company-id", "") if request else ""
    return [User.company_id == company_id]

Parameters:

Name	Type	Description	Default
data	Any	User authentication data transfer object.	required
request	Request | None	The litestar request that initiated the action.	None

Returns:

Type	Description
Sequence[ColumnElement[bool]]	A list of SQLAlchemy filter expressions to apply to the authentication query.

Notes: - Override this method in a subclass of BaseUserService to supply additional SQLAlchemy filter expressions that will be applied when looking up a user during authentication. - By default, no additional filters are applied.

generate_token

generate_token(user_id, aud)

Generate a limited time valid JWT.

Parameters:

Name	Type	Description	Default
user_id	UUID | int	ID of the user to provide the token to.	required
aud	str	Context of the token	required

initiate_verification async

initiate_verification(user)

Initiate the user verification flow.

Parameters:

Name	Type	Description	Default
user	SQLAUserT	The user requesting verification.	required

Notes

• The user verification flow is not initiated when require_verification_on_registration is set to False.

send_verification_token async

send_verification_token(user, token)

Execute custom logic to send the verification token to the relevant user.

Parameters:

Name	Type	Description	Default
user	SQLAUserT	The user requesting verification.	required
token	str	An encoded JWT bound to verification.	required

Notes: - Develepors need to override this method to facilitate sending the token via email, sms etc. - This method is not invoked when require_verification_on_registration is set to False.

verify async

verify(encoded_token, request=None)

Verify a user with the given JWT.

Parameters:

Name	Type	Description	Default
encoded_token	str	An encoded JWT bound to verification.	required
request	Request | None	The litestar request that initiated the action.	None

Raises:

Type	Description
InvalidTokenException	If the token is expired or tampered with.

initiate_password_reset async

initiate_password_reset(email)

Initiate the password reset flow.

Parameters:

Name	Type	Description	Default
email	str	Email of the user who has forgotten their password.	required

send_password_reset_token async

send_password_reset_token(user, token)

Execute custom logic to send the password reset token to the relevant user.

Parameters:

Name	Type	Description	Default
user	SQLAUserT	The user requesting the password reset.	required
token	str	An encoded JWT bound to the password reset flow.	required

Notes: - Develepors need to override this method to facilitate sending the token via email, sms etc.

reset_password async

reset_password(encoded_token, password)

Reset a user's password given a valid JWT.

Parameters:

Name	Type	Description	Default
encoded_token	str	An encoded JWT bound to the password reset flow.	required
password	str	The new password to hash and store.	required

Raises:

Type	Description
InvalidTokenException	If the token has expired or been tampered with.

pre_login_hook async

pre_login_hook(data, request=None)

Execute custom logic to run custom business logic prior to authenticating a user.

Useful for authentication checks against external sources, eg. current membership validity or blacklists, etc Must return False or raise a custom exception to cancel authentication.

Parameters:

Name	Type	Description	Default
data	AuthenticationSchema	Authentication data transfer object.	required
request	Request | None	The litestar request that initiated the action.	None

Notes

Uncaught exceptions in this method will break the authentication process.

post_login_hook async

post_login_hook(user, request=None)

Execute custom logic to run custom business logic after authenticating a user.

Useful for eg. updating a login counter, updating last known user IP address, etc.

Parameters:

Name	Type	Description	Default
user	SQLAUserT	The user who has authenticated.	required
request	Request | None	The litestar request that initiated the action.	None

Notes

Uncaught exceptions in this method will break the authentication process.

pre_registration_hook async

pre_registration_hook(data, request=None)

Execute custom logic to run custom business logic prior to registering a user.

Useful for authorization checks against external sources, eg. membership API or blacklists, etc.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	User creation data transfer object	required
request	Request | None	The litestar request that initiated the action.	None

Notes: - Uncaught exceptions in this method will result in failed registration attempts.

post_registration_hook async

post_registration_hook(user, request=None)

Execute custom logic to run custom business logic after registering a user.

Useful for updating external datasets, sending welcome messages etc.

Parameters:

Name	Type	Description	Default
user	SQLAUserT	User ORM instance.	required
request	Request | None	The litestar request that initiated the action.	None

Notes: - Uncaught exceptions in this method could result in returning a HTTP 500 status code while successfully creating the user in the database.

post_verification_hook async

post_verification_hook(user, request=None)

Execute custom logic to run custom business logic after a user has verified details.

Useful for eg. updating sales lead data, etc.

Parameters:

Name	Type	Description	Default
user	SQLAUserT	User ORM instance.	required
request	Request | None	The litestar request that initiated the action.	None

Notes: - Uncaught exceptions in this method could result in returning a HTTP 500 status code while successfully validating the user.

get_role async

get_role(role_id)

Retrieve a role by id.

Parameters:

Name	Type	Description	Default
role_id	UUID | int	ID of the role.	required

get_role_by_name async

get_role_by_name(name)

Retrieve a role by name.

Parameters:

Name	Type	Description	Default
name	str	The name of the role.	required

add_role async

add_role(data)

Add a new role to the database.

Parameters:

Name	Type	Description	Default
data	SQLARoleT	A role creation data transfer object.	required

update_role async

update_role(data)

Update a role in the database.

Parameters:

Name	Type	Description	Default
data	SQLARoleT	A role update data transfer object.	required

delete_role async

delete_role(role_id)

Delete a role from the database.

Parameters:

Name	Type	Description	Default
role_id	UUID | int	UUID corresponding to the role primary key.	required

assign_role async

assign_role(user_id, role_id)

Add a role to a user.

Parameters:

Name	Type	Description	Default
user_id	UUID | int	ID of the user to receive the role.	required
role_id	UUID | int	ID of the role to add to the user.	required

revoke_role async

revoke_role(user_id, role_id)

Revoke a role from a user.

Parameters:

Name	Type	Description	Default
user_id	UUID | int	ID of the user to revoke the role from.	required
role_id	UUID | int	ID of the role to revoke.	required

get_by_oauth_account async

get_by_oauth_account(oauth, account_id, request=None)

Get a user by OAuth account.

Parameters:

Name	Type	Description	Default
oauth	str	Name of the OAuth client.	required
account_id	str	Id. of the account on the external OAuth service.	required
request	Request | None	The litestar request that initiated the action.	None

Raises:

Type	Description
NotFoundError	The user or OAuth account does not exist.

Returns:

Type	Description
SQLAUserT	A user.

oauth2_authorize async

oauth2_authorize(request, oauth_client, state_secret, callback_route_name, scopes=None, redirect_url=None, state_data=None)

oauth2_callback async

oauth2_callback(data, oauth_client, state_secret, callback_route_name, request, redirect_url=None, associate_by_email=False, is_verified_by_default=False, is_associate_callback=False, associate_user=None)

Handle the callback after a successful OAuth authentication.

If the user already exists with this OAuth account, the token is updated.

If a user with the same e-mail already exists and associate_by_email is True, the OAuth account is associated to this user. Otherwise, the UserNotExists exception is raised.

If the user does not exist, it is created and the on_after_register handler is triggered.

:param oauth_name: Name of the OAuth client. :param access_token: Valid access token for the service provider. :param account_id: models.ID of the user on the service provider. :param account_email: E-mail of the user on the service provider. :param expires_at: Optional timestamp at which the access token expires. :param refresh_token: Optional refresh token to get a fresh access token from the service provider. :param request: Optional FastAPI request that triggered the operation, defaults to None :param associate_by_email: If True, any existing user with the same e-mail address will be associated to this user. Defaults to False. :param is_verified_by_default: If True, the is_verified flag will be set to True on newly created user. Make sure the OAuth Provider you're using does verify the email address before enabling this flag. Defaults to False. :param is_associate_callback: If True, the callback is an associate callback. :param associate_user: The user to associate the OAuth account to. :return: A user.