services#

Services for the accounts domain.

class app.domain.accounts.services.UserService(**repo_kwargs: Any)[source]#

Bases: SQLAlchemyAsyncRepositoryService[User]

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

  • load – Set default relationships to be loaded

  • execution_options – Set default execution options

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

repository_type#

alias of UserRepository

__init__(**repo_kwargs: Any) → None[source]#

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

  • load – Set default relationships to be loaded

  • execution_options – Set default execution options

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

async create(data: ~typing.Dict[str, ~typing.Any] | ~app.db.models.user.User | ~msgspec.Struct | ~advanced_alchemy.service.typing.BaseModel | ~litestar.dto.data_structures.DTOData[~app.db.models.user.User], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, auto_refresh: bool | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>) → User[source]#

Create a new User and assign default Role.

async update(data: ModelDictT[User], item_id: Any | None = None, *, id_attribute: str | InstrumentedAttribute[Any] | None = None, attribute_names: Iterable[str] | None = None, with_for_update: bool | None = None, auto_commit: bool | None = None, auto_expunge: bool | None = None, auto_refresh: bool | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) → User[source]#

Wrap repository update operation.

Parameters:

  • data – Representation to be updated.

  • item_id – Identifier of item to be updated.

  • attribute_names – an iterable of attribute names to pass into the update method.

  • with_for_update – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT

  • auto_expunge – Remove object from session before returning. Defaults to SQLAlchemyAsyncRepository.auto_expunge.

  • auto_refresh – Refresh object from session before returning. Defaults to SQLAlchemyAsyncRepository.auto_refresh

  • auto_commit – Commit objects before returning. Defaults to SQLAlchemyAsyncRepository.auto_commit

  • id_attribute – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.

  • error_messages – An optional dictionary of templates to use for friendlier error messages to clients

  • load – Set default relationships to be loaded

  • execution_options – Set default execution options

Returns:

Updated representation.

async authenticate(username: str, password: bytes | str) → User[source]#

Authenticate a user.

Parameters:

  • username (str) – _description_

  • password (str | bytes) – _description_

Raises:

NotAuthorizedException – Raised when the user doesn’t exist, isn’t verified, or is not active.

Returns:

The user object

Return type:

User

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

Update stored user password.

This is only used when not used IAP authentication.

Parameters:

  • data (UserPasswordUpdate) – _description_

  • db_obj (User) – _description_

Raises:

PermissionDeniedException – _description_

async static has_role_id(db_obj: User, role_id: UUID) → bool[source]#

Return true if user has specified role ID

async static has_role(db_obj: User, role_name: str) → bool[source]#

Return true if user has specified role ID

async to_model(data: Dict[str, Any] | User | Struct | BaseModel | DTOData[User], operation: str | None = None) → User[source]#

Parse and Convert input into a model.

Parameters:

  • data – Representations to be created.

  • operation – Optional operation flag so that you can provide behavior based on CRUD operation

Returns:

Representation of created instances.

class app.domain.accounts.services.RoleService(**repo_kwargs: Any)[source]#

Bases: SQLAlchemyAsyncRepositoryService[Role]

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

  • load – Set default relationships to be loaded

  • execution_options – Set default execution options

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

repository_type#

alias of RoleRepository

__init__(**repo_kwargs: Any) → None[source]#

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

  • load – Set default relationships to be loaded

  • execution_options – Set default execution options

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

async to_model(data: Dict[str, Any] | Role | Struct | BaseModel | DTOData[Role], operation: str | None = None) → Role[source]#

Parse and Convert input into a model.

Parameters:

  • data – Representations to be created.

  • operation – Optional operation flag so that you can provide behavior based on CRUD operation

Returns:

Representation of created instances.

class app.domain.accounts.services.UserRoleService(session: AsyncSession | async_scoped_session[AsyncSession], statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: list[OrderingPair] | OrderingPair | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None, **repo_kwargs: Any)[source]#

Bases: SQLAlchemyAsyncRepositoryService[UserRole]

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

  • load – Set default relationships to be loaded

  • execution_options – Set default execution options

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

repository_type#

alias of UserRoleRepository

class app.domain.accounts.services.UserOAuthAccountService(session: AsyncSession | async_scoped_session[AsyncSession], statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: list[OrderingPair] | OrderingPair | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None, **repo_kwargs: Any)[source]#

Bases: SQLAlchemyAsyncRepositoryService[UserOauthAccount]

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

  • load – Set default relationships to be loaded

  • execution_options – Set default execution options

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

repository_type#

alias of UserOauthAccountRepository

On this page
Edit on GitHub
Show Source