services

Services for the teams domain.

class app.domain.teams.services.TeamInvitationService(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[TeamInvitation, Any]

Team Invitation Service.

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 TeamInvitationRepository(*, 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[TeamInvitation]

Team Invitation 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 TeamInvitation

repository_type

alias of TeamInvitationRepository

class app.domain.teams.services.TeamMemberService(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[TeamMember, Any]

Team Member Service.

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 TeamMemberRepository(*, 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[TeamMember]

Team Member 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 TeamMember

repository_type

alias of TeamMemberRepository

class app.domain.teams.services.TeamService(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[Team, Any]

Team Service.

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 TeamRepository(*, 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[Team]

Team 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 Team

repository_type

alias of TeamRepository

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.Team]) → ModelDictT[m.Team]

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.Team]) → ModelDictT[m.Team]

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.Team]) → ModelDictT[m.Team]

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.