services

Services for the accounts domain.

Accounts domain services.

class app.domain.accounts.services.EmailVerificationTokenService(session: ~sqlalchemy.ext.asyncio.session.AsyncSession | ~sqlalchemy.ext.asyncio.scoping.async_scoped_session[~sqlalchemy.ext.asyncio.session.AsyncSession], *, statement: ~sqlalchemy.sql.selectable.Select | None = None, auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: list[tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | type[~advanced_alchemy.utils.dataclass.Empty] | None = <class 'advanced_alchemy.utils.dataclass.Empty'>, wrap_exceptions: bool = True, load: ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None, uniquify: bool | None = None, count_with_window_function: bool | None = None, **repo_kwargs: ~typing.Any)

Bases: SQLAlchemyAsyncRepositoryService[EmailVerificationToken, Any]

Handles database operations for email verification tokens.

Configure the service object.

Parameters:

• session – Session managing the unit-of-work for the operation.

• statement – To facilitate customization of the underlying select query.

• auto_expunge – Remove object from session before returning.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

• order_by – Set default order options for queries.

• error_messages – A set of custom error messages to use for operations

• wrap_exceptions – Wrap exceptions in a RepositoryError

• load – Set default relationships to be loaded

• execution_options – Set default execution options

• uniquify – Optionally apply the unique() method to results before returning.

• count_with_window_function – When false, list and count will use two queries instead of an analytical window function.

• **repo_kwargs – passed as keyword args to repo instantiation.

class Repo(*, statement: ~sqlalchemy.sql.selectable.Select | None = None, session: ~sqlalchemy.ext.asyncio.session.AsyncSession | ~sqlalchemy.ext.asyncio.scoping.async_scoped_session[~sqlalchemy.ext.asyncio.session.AsyncSession], auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: list[tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | type[~advanced_alchemy.utils.dataclass.Empty] | None = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None, wrap_exceptions: bool = True, uniquify: bool | None = None, count_with_window_function: bool | None = None, **kwargs: ~typing.Any)

Bases: SQLAlchemyAsyncRepository[EmailVerificationToken]

EmailVerificationToken SQLAlchemy Repository.

Repository for SQLAlchemy models.

Parameters:

• statement – To facilitate customization of the underlying select query.

• session – Session managing the unit-of-work for the operation.

• auto_expunge – Remove object from session before returning.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

• order_by – Set default order options for queries.

• load – Set default relationships to be loaded

• execution_options – Set default execution options

• error_messages – A set of custom error messages to use for operations

• wrap_exceptions – Wrap SQLAlchemy exceptions in a RepositoryError. When set to False, the original exception will be raised.

• uniquify – Optionally apply the unique() method to results before returning.

• count_with_window_function – When false, list and count will use two queries instead of an analytical window function.

• **kwargs – Additional arguments.

model_type

alias of EmailVerificationToken

async cleanup_expired_tokens() → int

Remove expired tokens from the database.

Returns:

Number of tokens removed

async create_verification_token(user_id: UUID, email: str) → tuple[m.EmailVerificationToken, str]

Create a new email verification token for a user.

Parameters:

• user_id – The user’s UUID

• email – The email address to verify

Returns:

Tuple of (EmailVerificationToken, plain_token)

async invalidate_user_tokens(user_id: UUID, email: str | None = None) → None

Invalidate all tokens for a user, optionally filtered by email.

Parameters:

• user_id – The user’s UUID

• email – Optional email to filter tokens

match_fields: ClassVar[list[str] | str | None] = ['token_hash']

List of dialects that prefer to use field.id = ANY(:1) instead of field.id IN (...).

repository_type

alias of Repo

async to_model_on_create(data: service.ModelDictT[m.EmailVerificationToken]) → service.ModelDictT[m.EmailVerificationToken]

Convenience method to allow for custom behavior on create.

Parameters:

data – The data to be converted to a model.

Returns:

The data to be converted to a model.

async verify_token(token: str) → EmailVerificationToken

Verify a token and mark it as used.

Parameters:

token – The verification token string

Returns:

The EmailVerificationToken instance if valid

Raises:

ClientException – If token is invalid, expired, or already used

class app.domain.accounts.services.PasswordResetService(session: ~sqlalchemy.ext.asyncio.session.AsyncSession | ~sqlalchemy.ext.asyncio.scoping.async_scoped_session[~sqlalchemy.ext.asyncio.session.AsyncSession], *, statement: ~sqlalchemy.sql.selectable.Select | None = None, auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: list[tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | type[~advanced_alchemy.utils.dataclass.Empty] | None = <class 'advanced_alchemy.utils.dataclass.Empty'>, wrap_exceptions: bool = True, load: ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None, uniquify: bool | None = None, count_with_window_function: bool | None = None, **repo_kwargs: ~typing.Any)

Bases: SQLAlchemyAsyncRepositoryService[PasswordResetToken, Any]

Handles database operations for password reset tokens.

Configure the service object.

Parameters:

• session – Session managing the unit-of-work for the operation.

• statement – To facilitate customization of the underlying select query.

• auto_expunge – Remove object from session before returning.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

• order_by – Set default order options for queries.

• error_messages – A set of custom error messages to use for operations

• wrap_exceptions – Wrap exceptions in a RepositoryError

• load – Set default relationships to be loaded

• execution_options – Set default execution options

• uniquify – Optionally apply the unique() method to results before returning.

• count_with_window_function – When false, list and count will use two queries instead of an analytical window function.

• **repo_kwargs – passed as keyword args to repo instantiation.

class Repo(*, statement: ~sqlalchemy.sql.selectable.Select | None = None, session: ~sqlalchemy.ext.asyncio.session.AsyncSession | ~sqlalchemy.ext.asyncio.scoping.async_scoped_session[~sqlalchemy.ext.asyncio.session.AsyncSession], auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: list[tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | type[~advanced_alchemy.utils.dataclass.Empty] | None = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None, wrap_exceptions: bool = True, uniquify: bool | None = None, count_with_window_function: bool | None = None, **kwargs: ~typing.Any)

Bases: SQLAlchemyAsyncRepository[PasswordResetToken]

PasswordResetToken SQLAlchemy Repository.

Repository for SQLAlchemy models.

Parameters:

• statement – To facilitate customization of the underlying select query.

• session – Session managing the unit-of-work for the operation.

• auto_expunge – Remove object from session before returning.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

• order_by – Set default order options for queries.

• load – Set default relationships to be loaded

• execution_options – Set default execution options

• error_messages – A set of custom error messages to use for operations

• wrap_exceptions – Wrap SQLAlchemy exceptions in a RepositoryError. When set to False, the original exception will be raised.

• uniquify – Optionally apply the unique() method to results before returning.

• count_with_window_function – When false, list and count will use two queries instead of an analytical window function.

• **kwargs – Additional arguments.

model_type

alias of PasswordResetToken

async check_rate_limit(user_id: UUID, hours: float = 1) → bool

Check if user has exceeded reset token creation rate limit.

Parameters:

• user_id – The user’s UUID

• hours – Hours to look back for rate limiting

Returns:

True if rate limit exceeded, False otherwise

async cleanup_expired_tokens() → int

Remove expired tokens from the database.

Returns:

Number of tokens removed

async create_reset_token(user_id: UUID, ip_address: str | None = None, user_agent: str | None = None) → tuple[m.PasswordResetToken, str]

Create a new password reset token for a user.

Parameters:

• user_id – The user’s UUID

• ip_address – IP address of the request

• user_agent – User agent string of the request

Returns:

Tuple of (PasswordResetToken, plain_token)

async invalidate_user_tokens(user_id: UUID) → None

Invalidate all active tokens for a user.

Parameters:

user_id – The user’s UUID

match_fields: ClassVar[list[str] | str | None] = ['token_hash']

List of dialects that prefer to use field.id = ANY(:1) instead of field.id IN (...).

repository_type

alias of Repo

async to_model_on_create(data: service.ModelDictT[m.PasswordResetToken]) → service.ModelDictT[m.PasswordResetToken]

Convenience method to allow for custom behavior on create.

Parameters:

data – The data to be converted to a model.

Returns:

The data to be converted to a model.

async use_reset_token(token: str) → PasswordResetToken

Use a token to mark it as consumed.

Parameters:

token – The reset token string

Returns:

The PasswordResetToken instance

async validate_reset_token(token: str) → PasswordResetToken

Validate a token without consuming it.

Parameters:

token – The reset token string

Returns:

The PasswordResetToken instance if valid.

Raises:

ClientException – If token is invalid, expired, or already used

class app.domain.accounts.services.RefreshTokenService(session: ~sqlalchemy.ext.asyncio.session.AsyncSession | ~sqlalchemy.ext.asyncio.scoping.async_scoped_session[~sqlalchemy.ext.asyncio.session.AsyncSession], *, statement: ~sqlalchemy.sql.selectable.Select | None = None, auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: list[tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | type[~advanced_alchemy.utils.dataclass.Empty] | None = <class 'advanced_alchemy.utils.dataclass.Empty'>, wrap_exceptions: bool = True, load: ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None, uniquify: bool | None = None, count_with_window_function: bool | None = None, **repo_kwargs: ~typing.Any)

Bases: SQLAlchemyAsyncRepositoryService[RefreshToken, Any]

Handles database operations for refresh tokens with rotation and reuse detection.

Tokens are stored as SHA-256 hashes, never plaintext. Each token belongs to a ‘family’ for reuse detection - if a revoked token is presented, the entire family is revoked for security.

Configure the service object.

Parameters:

• session – Session managing the unit-of-work for the operation.

• statement – To facilitate customization of the underlying select query.

• auto_expunge – Remove object from session before returning.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

• order_by – Set default order options for queries.

• error_messages – A set of custom error messages to use for operations

• wrap_exceptions – Wrap exceptions in a RepositoryError

• load – Set default relationships to be loaded

• execution_options – Set default execution options

• uniquify – Optionally apply the unique() method to results before returning.

• count_with_window_function – When false, list and count will use two queries instead of an analytical window function.

• **repo_kwargs – passed as keyword args to repo instantiation.

class Repo(*, statement: ~sqlalchemy.sql.selectable.Select | None = None, session: ~sqlalchemy.ext.asyncio.session.AsyncSession | ~sqlalchemy.ext.asyncio.scoping.async_scoped_session[~sqlalchemy.ext.asyncio.session.AsyncSession], auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: list[tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | type[~advanced_alchemy.utils.dataclass.Empty] | None = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None, wrap_exceptions: bool = True, uniquify: bool | None = None, count_with_window_function: bool | None = None, **kwargs: ~typing.Any)

Bases: SQLAlchemyAsyncRepository[RefreshToken]

RefreshToken SQLAlchemy Repository.

Repository for SQLAlchemy models.

Parameters:

• statement – To facilitate customization of the underlying select query.

• session – Session managing the unit-of-work for the operation.

• auto_expunge – Remove object from session before returning.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

• order_by – Set default order options for queries.

• load – Set default relationships to be loaded

• execution_options – Set default execution options

• error_messages – A set of custom error messages to use for operations

• wrap_exceptions – Wrap SQLAlchemy exceptions in a RepositoryError. When set to False, the original exception will be raised.

• uniquify – Optionally apply the unique() method to results before returning.

• count_with_window_function – When false, list and count will use two queries instead of an analytical window function.

• **kwargs – Additional arguments.

model_type

alias of RefreshToken

async cleanup_expired_tokens() → int

Remove expired and old revoked tokens from the database.

This should be run periodically as a background job.

Returns:

Number of tokens removed

async create_refresh_token(user_id: UUID, family_id: UUID | None = None, device_info: str | None = None, expiration_days: int = 7) → tuple[str, m.RefreshToken]

Create a new refresh token for a user.

Parameters:

• user_id – The user’s UUID

• family_id – Optional family ID for token rotation (creates new if None)

• device_info – Optional device fingerprint (user agent, etc.)

• expiration_days – Number of days until token expires

Returns:

Tuple of (raw_token, RefreshToken instance) The raw_token should be sent to the client, the model is stored in DB.

async get_active_sessions(user_id: UUID) → list[m.RefreshToken]

Get all active refresh tokens for a user.

Useful for showing active sessions in account settings.

Parameters:

user_id – The user’s UUID

Returns:

List of active RefreshToken instances

static hash_token(token: str) → str

Hash a token using SHA-256.

Parameters:

token – The raw token string

Returns:

SHA-256 hex digest of the token

match_fields: ClassVar[list[str] | str | None] = ['token_hash']

List of dialects that prefer to use field.id = ANY(:1) instead of field.id IN (...).

repository_type

alias of Repo

async revoke_token_family(family_id: UUID) → int

Revoke all tokens in a family.

Used for logout and security purposes (reuse detection).

Parameters:

family_id – The family ID to revoke

Returns:

Number of tokens revoked

async revoke_user_tokens(user_id: UUID) → int

Revoke all refresh tokens for a user.

Used for password changes, security events, or explicit logout from all devices.

Parameters:

user_id – The user’s UUID

Returns:

Number of tokens revoked

async rotate_refresh_token(raw_token: str, device_info: str | None = None) → tuple[str, RefreshToken]

Rotate a refresh token, creating a new one and revoking the old.

This implements refresh token rotation with reuse detection. If a revoked token is presented, the entire family is revoked.

Parameters:

• raw_token – The raw token string from the client

• device_info – Optional device fingerprint

Returns:

Tuple of (new_raw_token, new_RefreshToken)

async validate_refresh_token(raw_token: str) → RefreshToken

Validate a refresh token without consuming it.

Parameters:

raw_token – The raw token string from the client

Returns:

The RefreshToken instance if valid.

Raises:

NotAuthorizedException – If token is invalid, expired, or revoked

class app.domain.accounts.services.RoleService(session: ~sqlalchemy.ext.asyncio.session.AsyncSession | ~sqlalchemy.ext.asyncio.scoping.async_scoped_session[~sqlalchemy.ext.asyncio.session.AsyncSession], *, statement: ~sqlalchemy.sql.selectable.Select | None = None, auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: list[tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | type[~advanced_alchemy.utils.dataclass.Empty] | None = <class 'advanced_alchemy.utils.dataclass.Empty'>, wrap_exceptions: bool = True, load: ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None, uniquify: bool | None = None, count_with_window_function: bool | None = None, **repo_kwargs: ~typing.Any)

Bases: SQLAlchemyAsyncRepositoryService[Role, Any]

Handles database operations for users.

Configure the service object.

Parameters:

• session – Session managing the unit-of-work for the operation.

• statement – To facilitate customization of the underlying select query.

• auto_expunge – Remove object from session before returning.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

• order_by – Set default order options for queries.

• error_messages – A set of custom error messages to use for operations

• wrap_exceptions – Wrap exceptions in a RepositoryError

• load – Set default relationships to be loaded

• execution_options – Set default execution options

• uniquify – Optionally apply the unique() method to results before returning.

• count_with_window_function – When false, list and count will use two queries instead of an analytical window function.

• **repo_kwargs – passed as keyword args to repo instantiation.

class Repo(*, statement: ~sqlalchemy.sql.selectable.Select | None = None, session: ~sqlalchemy.ext.asyncio.session.AsyncSession | ~sqlalchemy.ext.asyncio.scoping.async_scoped_session[~sqlalchemy.ext.asyncio.session.AsyncSession], auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: list[tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | type[~advanced_alchemy.utils.dataclass.Empty] | None = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None, wrap_exceptions: bool = True, uniquify: bool | None = None, count_with_window_function: bool | None = None, **kwargs: ~typing.Any)

Bases: SQLAlchemyAsyncSlugRepository[Role]

User SQLAlchemy Repository.

Repository for SQLAlchemy models.

Parameters:

• statement – To facilitate customization of the underlying select query.

• session – Session managing the unit-of-work for the operation.

• auto_expunge – Remove object from session before returning.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

• order_by – Set default order options for queries.

• load – Set default relationships to be loaded

• execution_options – Set default execution options

• error_messages – A set of custom error messages to use for operations

• wrap_exceptions – Wrap SQLAlchemy exceptions in a RepositoryError. When set to False, the original exception will be raised.

• uniquify – Optionally apply the unique() method to results before returning.

• count_with_window_function – When false, list and count will use two queries instead of an analytical window function.

• **kwargs – Additional arguments.

model_type

alias of Role

match_fields: ClassVar[list[str] | str | None] = ['name']

List of dialects that prefer to use field.id = ANY(:1) instead of field.id IN (...).

repository_type

alias of Repo

async to_model_on_create(data: service.ModelDictT[m.Role]) → service.ModelDictT[m.Role]

Convenience method to allow for custom behavior on create.

Parameters:

data – The data to be converted to a model.

Returns:

The data to be converted to a model.

async to_model_on_update(data: service.ModelDictT[m.Role]) → service.ModelDictT[m.Role]

Convenience method to allow for custom behavior on update.

Parameters:

data – The data to be converted to a model.

Returns:

The data to be converted to a model.

async to_model_on_upsert(data: service.ModelDictT[m.Role]) → service.ModelDictT[m.Role]

Convenience method to allow for custom behavior on upsert.

Parameters:

data – The data to be converted to a model.

Returns:

The data to be converted to a model.

class app.domain.accounts.services.UserOAuthAccountService(session: ~sqlalchemy.ext.asyncio.session.AsyncSession | ~sqlalchemy.ext.asyncio.scoping.async_scoped_session[~sqlalchemy.ext.asyncio.session.AsyncSession], *, statement: ~sqlalchemy.sql.selectable.Select | None = None, auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: list[tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | type[~advanced_alchemy.utils.dataclass.Empty] | None = <class 'advanced_alchemy.utils.dataclass.Empty'>, wrap_exceptions: bool = True, load: ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None, uniquify: bool | None = None, count_with_window_function: bool | None = None, **repo_kwargs: ~typing.Any)

Bases: SQLAlchemyAsyncRepositoryService[UserOAuthAccount, Any]

Handles database operations for user OAuth external authorization.

Configure the service object.

Parameters:

• session – Session managing the unit-of-work for the operation.

• statement – To facilitate customization of the underlying select query.

• auto_expunge – Remove object from session before returning.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

• order_by – Set default order options for queries.

• error_messages – A set of custom error messages to use for operations

• wrap_exceptions – Wrap exceptions in a RepositoryError

• load – Set default relationships to be loaded

• execution_options – Set default execution options

• uniquify – Optionally apply the unique() method to results before returning.

• count_with_window_function – When false, list and count will use two queries instead of an analytical window function.

• **repo_kwargs – passed as keyword args to repo instantiation.

class Repo(*, statement: ~sqlalchemy.sql.selectable.Select | None = None, session: ~sqlalchemy.ext.asyncio.session.AsyncSession | ~sqlalchemy.ext.asyncio.scoping.async_scoped_session[~sqlalchemy.ext.asyncio.session.AsyncSession], auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: list[tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | type[~advanced_alchemy.utils.dataclass.Empty] | None = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None, wrap_exceptions: bool = True, uniquify: bool | None = None, count_with_window_function: bool | None = None, **kwargs: ~typing.Any)

Bases: SQLAlchemyAsyncRepository[UserOAuthAccount]

User SQLAlchemy Repository.

Repository for SQLAlchemy models.

Parameters:

• statement – To facilitate customization of the underlying select query.

• session – Session managing the unit-of-work for the operation.

• auto_expunge – Remove object from session before returning.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

• order_by – Set default order options for queries.

• load – Set default relationships to be loaded

• execution_options – Set default execution options

• error_messages – A set of custom error messages to use for operations

• wrap_exceptions – Wrap SQLAlchemy exceptions in a RepositoryError. When set to False, the original exception will be raised.

• uniquify – Optionally apply the unique() method to results before returning.

• count_with_window_function – When false, list and count will use two queries instead of an analytical window function.

• **kwargs – Additional arguments.

model_type

alias of UserOAuthAccount

async can_unlink_oauth(user: User) → tuple[bool, str]

Check if user can safely unlink an OAuth provider.

Parameters:

user – The user attempting to unlink.

Returns:

Tuple of (can_unlink, reason_if_not).

async create_or_update_oauth_account(user_id: UUID, provider: str, oauth_data: dict[str, Any], token_data: OAuth2Token) → m.UserOAuthAccount

Create or update OAuth account with token data.

async get_by_provider_account_id(provider: str, account_id: str) → UserOAuthAccount | None

Get an OAuth account by provider and account ID.

async link_or_update_oauth(user_id: UUID, provider: str, account_id: str, account_email: str | None, access_token: str, refresh_token: str | None = None, expires_at: int | None = None, scopes: list[str] | str | None = None, provider_user_data: dict[str, Any] | None = None, last_login_at: datetime | None = None) → m.UserOAuthAccount

Link a new OAuth account or update existing one.

repository_type

alias of Repo

async unlink_oauth_account(user_id: UUID, provider: str) → bool

Unlink OAuth account from user.

class app.domain.accounts.services.UserRoleService(session: ~sqlalchemy.ext.asyncio.session.AsyncSession | ~sqlalchemy.ext.asyncio.scoping.async_scoped_session[~sqlalchemy.ext.asyncio.session.AsyncSession], *, statement: ~sqlalchemy.sql.selectable.Select | None = None, auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: list[tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | type[~advanced_alchemy.utils.dataclass.Empty] | None = <class 'advanced_alchemy.utils.dataclass.Empty'>, wrap_exceptions: bool = True, load: ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None, uniquify: bool | None = None, count_with_window_function: bool | None = None, **repo_kwargs: ~typing.Any)

Bases: SQLAlchemyAsyncRepositoryService[UserRole, Any]

Handles database operations for user roles.

Configure the service object.

Parameters:

• session – Session managing the unit-of-work for the operation.

• statement – To facilitate customization of the underlying select query.

• auto_expunge – Remove object from session before returning.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

• order_by – Set default order options for queries.

• error_messages – A set of custom error messages to use for operations

• wrap_exceptions – Wrap exceptions in a RepositoryError

• load – Set default relationships to be loaded

• execution_options – Set default execution options

• uniquify – Optionally apply the unique() method to results before returning.

• count_with_window_function – When false, list and count will use two queries instead of an analytical window function.

• **repo_kwargs – passed as keyword args to repo instantiation.

class Repo(*, statement: ~sqlalchemy.sql.selectable.Select | None = None, session: ~sqlalchemy.ext.asyncio.session.AsyncSession | ~sqlalchemy.ext.asyncio.scoping.async_scoped_session[~sqlalchemy.ext.asyncio.session.AsyncSession], auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: list[tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | type[~advanced_alchemy.utils.dataclass.Empty] | None = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None, wrap_exceptions: bool = True, uniquify: bool | None = None, count_with_window_function: bool | None = None, **kwargs: ~typing.Any)

Bases: SQLAlchemyAsyncRepository[UserRole]

User Role SQLAlchemy Repository.

Repository for SQLAlchemy models.

Parameters:

• statement – To facilitate customization of the underlying select query.

• session – Session managing the unit-of-work for the operation.

• auto_expunge – Remove object from session before returning.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

• order_by – Set default order options for queries.

• load – Set default relationships to be loaded

• execution_options – Set default execution options

• error_messages – A set of custom error messages to use for operations

• wrap_exceptions – Wrap SQLAlchemy exceptions in a RepositoryError. When set to False, the original exception will be raised.

• uniquify – Optionally apply the unique() method to results before returning.

• count_with_window_function – When false, list and count will use two queries instead of an analytical window function.

• **kwargs – Additional arguments.

model_type

alias of UserRole

repository_type

alias of Repo

class app.domain.accounts.services.UserService(session: ~sqlalchemy.ext.asyncio.session.AsyncSession | ~sqlalchemy.ext.asyncio.scoping.async_scoped_session[~sqlalchemy.ext.asyncio.session.AsyncSession], *, statement: ~sqlalchemy.sql.selectable.Select | None = None, auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: list[tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | type[~advanced_alchemy.utils.dataclass.Empty] | None = <class 'advanced_alchemy.utils.dataclass.Empty'>, wrap_exceptions: bool = True, load: ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None, uniquify: bool | None = None, count_with_window_function: bool | None = None, **repo_kwargs: ~typing.Any)

Bases: CompositeServiceMixin, SQLAlchemyAsyncRepositoryService[User, Any]

Handles database operations for users.

Configure the service object.

Parameters:

• session – Session managing the unit-of-work for the operation.

• statement – To facilitate customization of the underlying select query.

• auto_expunge – Remove object from session before returning.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

• order_by – Set default order options for queries.

• error_messages – A set of custom error messages to use for operations

• wrap_exceptions – Wrap exceptions in a RepositoryError

• load – Set default relationships to be loaded

• execution_options – Set default execution options

• uniquify – Optionally apply the unique() method to results before returning.

• count_with_window_function – When false, list and count will use two queries instead of an analytical window function.

• **repo_kwargs – passed as keyword args to repo instantiation.

class Repo(*, statement: ~sqlalchemy.sql.selectable.Select | None = None, session: ~sqlalchemy.ext.asyncio.session.AsyncSession | ~sqlalchemy.ext.asyncio.scoping.async_scoped_session[~sqlalchemy.ext.asyncio.session.AsyncSession], auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: list[tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | type[~advanced_alchemy.utils.dataclass.Empty] | None = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None, wrap_exceptions: bool = True, uniquify: bool | None = None, count_with_window_function: bool | None = None, **kwargs: ~typing.Any)

Bases: SQLAlchemyAsyncRepository[User]

User SQLAlchemy Repository.

Repository for SQLAlchemy models.

Parameters:

• statement – To facilitate customization of the underlying select query.

• session – Session managing the unit-of-work for the operation.

• auto_expunge – Remove object from session before returning.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

• order_by – Set default order options for queries.

• load – Set default relationships to be loaded

• execution_options – Set default execution options

• error_messages – A set of custom error messages to use for operations

• wrap_exceptions – Wrap SQLAlchemy exceptions in a RepositoryError. When set to False, the original exception will be raised.

• uniquify – Optionally apply the unique() method to results before returning.

• count_with_window_function – When false, list and count will use two queries instead of an analytical window function.

• **kwargs – Additional arguments.

model_type

alias of User

async authenticate(username: str, password: bytes | str) → User

Authenticate a user against the stored hashed password.

Returns:

The user object if authentication is successful.

Raises:

PermissionDeniedException – If the user is not found, the password is invalid, or the account is inactive.

async authenticate_or_create_oauth_user(provider: str, oauth_data: dict[str, Any], token_data: OAuth2Token) → tuple[m.User, bool]

Authenticate existing OAuth user or create new one.

Parameters:

• provider – OAuth provider name

• oauth_data – User data from OAuth provider

• token_data – OAuth token data

Returns:

Tuple of (user, is_new_user)

async create_user_from_oauth(oauth_data: dict[str, Any], provider: str, token_data: OAuth2Token) → m.User

Create new user from OAuth data.

Parameters:

• oauth_data – User data from OAuth provider

• provider – OAuth provider name (e.g., ‘google’)

• token_data – OAuth token data

Returns:

The created user object

async static has_role(db_obj: User, role_name: str) → bool

Return true if user has specified role ID

async static has_role_id(db_obj: m.User, role_id: UUID) → bool

Return true if user has specified role ID

async increment_failed_reset_attempt(user_id: UUID) → None

Increment failed reset attempts counter.

Parameters:

user_id – The user’s UUID

async is_email_verified(user_id: UUID) → bool

Check if user’s email is verified.

Parameters:

user_id – The user’s UUID

Returns:

True if email is verified, False otherwise

async is_reset_rate_limited(user_id: UUID) → bool

Check if user is rate limited for password resets.

Parameters:

user_id – The user’s UUID

Returns:

True if user is rate limited, False otherwise

match_fields: ClassVar[list[str] | str | None] = ['email']

List of dialects that prefer to use field.id = ANY(:1) instead of field.id IN (...).

property oauth_accounts: UserOAuthAccountService

Lazy-loaded OAuth account service sharing this session.

repository_type

alias of Repo

async require_verified_email(user: User) → None

Raise exception if user’s email is not verified.

Parameters:

user – The user object to check

Raises:

PermissionDeniedException – If email is not verified

async reset_password_with_token(user_id: UUID, new_password: str) → m.User

Reset user’s password using a validated token.

Parameters:

• user_id – The user’s UUID

• new_password – The new password

Returns:

The updated user object

Raises:

ClientException – If user not found or password validation fails

async to_model_on_create(data: service.ModelDictT[m.User]) → service.ModelDictT[m.User]

Convenience method to allow for custom behavior on create.

Parameters:

data – The data to be converted to a model.

Returns:

The data to be converted to a model.

async to_model_on_update(data: service.ModelDictT[m.User]) → service.ModelDictT[m.User]

Convenience method to allow for custom behavior on update.

Parameters:

data – The data to be converted to a model.

Returns:

The data to be converted to a model.

async to_model_on_upsert(data: service.ModelDictT[m.User]) → service.ModelDictT[m.User]

Convenience method to allow for custom behavior on upsert.

Parameters:

data – The data to be converted to a model.

Returns:

The data to be converted to a model.

async update_password(data: dict[str, Any], db_obj: User) → None

Modify stored user password.

Raises:

PermissionDeniedException – If the user is not found, the password is invalid, or the account is inactive.

async verify_email(user_id: UUID, email: str) → m.User

Mark user’s email as verified.

Parameters:

• user_id – The user’s UUID

• email – The email address that was verified

Returns:

The updated user object

Raises:

ClientException – If user not found or email doesn’t match