services

Services for the accounts domain.

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]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | type[~advanced_alchemy.utils.dataclass.Empty] = <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[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 UserRepository(*, 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]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | 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

repository_type

alias of UserRepository

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

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

async to_model_on_create(data: ModelDictT[m.User]) → 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: ModelDictT[m.User]) → 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: ModelDictT[m.User]) → 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 authenticate(username: str, password: bytes | str) → User

Authenticate a user against the stored hashed password.

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

Modify stored user password.

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

Return true if user has specified role ID

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

Return true if user has specified role ID

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]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | type[~advanced_alchemy.utils.dataclass.Empty] = <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 Repository(*, 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]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | 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

repository_type

alias of Repository

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

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

async to_model_on_create(data: ModelDictT[m.Role]) → 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: ModelDictT[m.Role]) → 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.

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]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | type[~advanced_alchemy.utils.dataclass.Empty] = <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 Repository(*, 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]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | 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 Repository

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]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | type[~advanced_alchemy.utils.dataclass.Empty] = <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 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 Repository(*, 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]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | 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

repository_type

alias of Repository