services

class advanced_alchemy.service.SQLAlchemyAsyncRepositoryService

Bases: SQLAlchemyAsyncRepositoryReadService[ModelT]

Service object that operates on a repository object.

async create(data: ~typing.Dict[str, ~typing.Any] | ~advanced_alchemy.repository.typing.ModelT | ~msgspec.Struct | ~pydantic.main.BaseModel | ~litestar.dto.data_structures.DTOData[~advanced_alchemy.repository.typing.ModelT], *, 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'>) → ModelT

Wrap repository instance creation.

Parameters:

• data – Representation to be created.

• 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

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

Returns:

Representation of created instance.

async create_many(data: BulkModelDictT[ModelT], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>) → Sequence[ModelT]

Wrap repository bulk instance creation.

Parameters:

• data – Representations to be created.

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

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

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

Returns:

Representation of created instances.

async update(data: ModelDictT[ModelT], item_id: 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, id_attribute: str | InstrumentedAttribute[Any] | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) → ModelT

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 update_many(data: BulkModelDictT[ModelT], *, auto_commit: bool | None = None, auto_expunge: 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) → Sequence[ModelT]

Wrap repository bulk instance update.

Parameters:

• data – Representations to be updated.

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

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

• 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:

Representation of updated instances.

async upsert(data: ~typing.Dict[str, ~typing.Any] | ~advanced_alchemy.repository.typing.ModelT | ~msgspec.Struct | ~pydantic.main.BaseModel | ~litestar.dto.data_structures.DTOData[~advanced_alchemy.repository.typing.ModelT], item_id: ~typing.Any | None = None, *, attribute_names: ~typing.Iterable[str] | None = None, with_for_update: bool | None = None, auto_expunge: bool | None = None, auto_commit: bool | None = None, auto_refresh: bool | None = None, match_fields: list[str] | str | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.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] | ~typing.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 | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, typing.Any] | None = None) → ModelT

Wrap repository upsert operation.

Parameters:

• data – Instance to update existing, or be created. Identifier used to determine if an existing instance exists is the value of an attribute on data named as value of self.id_attribute.

• item_id – Identifier of the object for upsert.

• 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

• match_fields – a list of keys to use to match the existing model. When empty, all fields are matched.

• 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 or created representation.

async upsert_many(data: BulkModelDictT[ModelT], *, auto_expunge: bool | None = None, auto_commit: bool | None = None, no_merge: bool = False, match_fields: list[str] | str | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) → Sequence[ModelT]

Wrap repository upsert operation.

Parameters:

• data – Instance to update existing, or be created. Identifier used to determine if an existing instance exists is the value of an attribute on data named as value of id_attribute.

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

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

• no_merge – Skip the usage of optimized Merge statements SQLAlchemyAsyncRepository.auto_commit

• match_fields – a list of keys to use to match the existing model. When empty, all fields are matched.

• 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 or created representation.

async get_or_upsert(*filters: StatementFilter | ColumnElement[bool], match_fields: list[str] | str | None = None, upsert: bool = True, 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, **kwargs: Any) → tuple[ModelT, bool]

Wrap repository instance creation.

Parameters:

• *filters – Types for specific filtering operations.

• match_fields – a list of keys to use to match the existing model. When empty, all fields are matched.

• upsert – When using match_fields and actual model values differ from kwargs, perform an update operation on the model.

• create – Should a model be created. If no model is found, an exception is raised.

• 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

• 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

• **kwargs – Identifier of the instance to be retrieved.

Returns:

Representation of created instance.

async get_and_update(*filters: StatementFilter | ColumnElement[bool], match_fields: list[str] | str | 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, **kwargs: Any) → tuple[ModelT, bool]

Wrap repository instance creation.

Parameters:

• *filters – Types for specific filtering operations.

• match_fields – a list of keys to use to match the existing model. When empty, all fields are matched.

• 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

• 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

• **kwargs – Identifier of the instance to be retrieved.

Returns:

Representation of updated instance.

async delete(item_id: Any, *, auto_commit: bool | None = None, auto_expunge: bool | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) → ModelT

Wrap repository delete operation.

Parameters:

• item_id – Identifier of instance to be deleted.

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

• 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:

Representation of the deleted instance.

async delete_many(item_ids: list[Any], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, chunk_size: int | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) → Sequence[ModelT]

Wrap repository bulk instance deletion.

Parameters:

• item_ids – Identifier of instance to be deleted.

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

• 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.

• chunk_size – Allows customization of the insertmanyvalues_max_parameters setting for the driver. Defaults to 950 if left unset.

• 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:

Representation of removed instances.

async delete_where(*filters: StatementFilter | ColumnElement[bool], auto_commit: bool | None = None, auto_expunge: bool | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, sanity_check: bool = True, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None, **kwargs: Any) → Sequence[ModelT]

Wrap repository scalars operation.

Parameters:

• *filters – Types for specific filtering operations.

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

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

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

• sanity_check – When true, the length of selected instances is compared to the deleted row count

• load – Set default relationships to be loaded

• execution_options – Set default execution options

• **kwargs – Instance attribute value filters.

Returns:

The list of instances deleted from the repository.

class advanced_alchemy.service.SQLAlchemyAsyncQueryService

Bases: ResultConverter

Simple service to execute the basic Query repository..

__init__(session: AsyncSession | async_scoped_session[AsyncSession], **repo_kwargs: Any) → None

Configure the service object.

Parameters:

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

• **repo_kwargs – Optional configuration values to pass into the repository

classmethod new(session: AsyncSession | async_scoped_session[AsyncSession] | None = None, config: SQLAlchemyAsyncConfig | None = None) → AsyncIterator[Self]

Context manager that returns instance of service object.

Handles construction of the database session._create_select_for_model

Returns:

The service object instance.

class advanced_alchemy.service.SQLAlchemySyncQueryService

Bases: ResultConverter

Simple service to execute the basic Query repository..

__init__(session: Session | scoped_session[Session], **repo_kwargs: Any) → None

Configure the service object.

Parameters:

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

• **repo_kwargs – Optional configuration values to pass into the repository

classmethod new(session: Session | scoped_session[Session] | None = None, config: SQLAlchemySyncConfig | None = None) → Iterator[Self]

Context manager that returns instance of service object.

Handles construction of the database session._create_select_for_model

Returns:

The service object instance.

class advanced_alchemy.service.SQLAlchemySyncRepositoryReadService

Bases: Generic[ModelT], ResultConverter

Service object that operates on a repository object.

__init__(session: Session | scoped_session[Session], 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) → None

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.

count(*filters: StatementFilter | ColumnElement[bool], statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None, **kwargs: Any) → int

Count of records returned by query.

Parameters:

• *filters – arguments for filtering.

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

• **kwargs – key value pairs of filter types.

Returns:

A count of the collection, filtered, but ignoring pagination.

exists(*filters: StatementFilter | ColumnElement[bool], error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None, **kwargs: Any) → bool

Wrap repository exists operation.

Parameters:

• *filters – Types for specific filtering operations.

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

• **kwargs – Keyword arguments for attribute based filtering.

Returns:

Representation of instance with identifier item_id.

get(item_id: Any, *, statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, auto_expunge: 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) → ModelT

Wrap repository scalar operation.

Parameters:

• item_id – Identifier of instance to be retrieved.

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

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

• 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 relationships to be loaded

• execution_options – Set default execution options

Returns:

Representation of instance with identifier item_id.

get_one(*filters: StatementFilter | ColumnElement[bool], statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, auto_expunge: bool | None = None, load: LoadSpec | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, execution_options: dict[str, Any] | None = None, **kwargs: Any) → ModelT

Wrap repository scalar operation.

Parameters:

• *filters – Types for specific filtering operations.

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

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

• 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

• **kwargs – Identifier of the instance to be retrieved.

Returns:

Representation of instance with identifier item_id.

get_one_or_none(*filters: StatementFilter | ColumnElement[bool], statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, auto_expunge: 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, **kwargs: Any) → ModelT | None

Wrap repository scalar operation.

Parameters:

• *filters – Types for specific filtering operations.

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

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

• 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

• **kwargs – Identifier of the instance to be retrieved.

Returns:

Representation of instance with identifier item_id.

to_model(data: Dict[str, Any] | ModelT | Struct | BaseModel | DTOData[ModelT], operation: str | None = None) → ModelT

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.

list_and_count(*filters: StatementFilter | ColumnElement[bool], statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, auto_expunge: bool | None = None, force_basic_query_mode: bool | None = None, 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, **kwargs: Any) → tuple[Sequence[ModelT], int]

List of records and total count returned by query.

Parameters:

• *filters – Types for specific filtering operations.

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

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

• force_basic_query_mode – Force list and count to use two queries instead of an analytical window function.

• order_by – Set default order options for queries.

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

• **kwargs – Instance attribute value filters.

Returns:

List of instances and count of total collection, ignoring pagination.

classmethod new(session: Session | scoped_session[Session] | None = None, statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, config: SQLAlchemySyncConfig | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) → Iterator[Self]

Context manager that returns instance of service object.

Handles construction of the database session._create_select_for_model

Returns:

The service object instance.

list(*filters: StatementFilter | ColumnElement[bool], statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, auto_expunge: bool | None = None, 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, **kwargs: Any) → Sequence[ModelT]

Wrap repository scalars operation.

Parameters:

• *filters – Types for specific filtering operations.

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

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

• order_by – Set default order options for queries.

• 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

• **kwargs – Instance attribute value filters.

Returns:

The list of instances retrieved from the repository.

class advanced_alchemy.service.SQLAlchemySyncRepositoryService

Bases: SQLAlchemySyncRepositoryReadService[ModelT]

Service object that operates on a repository object.

create(data: ~typing.Dict[str, ~typing.Any] | ~advanced_alchemy.repository.typing.ModelT | ~msgspec.Struct | ~pydantic.main.BaseModel | ~litestar.dto.data_structures.DTOData[~advanced_alchemy.repository.typing.ModelT], *, 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'>) → ModelT

Wrap repository instance creation.

Parameters:

• data – Representation to be created.

• 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

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

Returns:

Representation of created instance.

create_many(data: BulkModelDictT[ModelT], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>) → Sequence[ModelT]

Wrap repository bulk instance creation.

Parameters:

• data – Representations to be created.

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

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

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

Returns:

Representation of created instances.

update(data: ModelDictT[ModelT], item_id: 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, id_attribute: str | InstrumentedAttribute[Any] | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) → ModelT

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.

update_many(data: BulkModelDictT[ModelT], *, auto_commit: bool | None = None, auto_expunge: 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) → Sequence[ModelT]

Wrap repository bulk instance update.

Parameters:

• data – Representations to be updated.

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

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

• 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:

Representation of updated instances.

upsert(data: ~typing.Dict[str, ~typing.Any] | ~advanced_alchemy.repository.typing.ModelT | ~msgspec.Struct | ~pydantic.main.BaseModel | ~litestar.dto.data_structures.DTOData[~advanced_alchemy.repository.typing.ModelT], item_id: ~typing.Any | None = None, *, attribute_names: ~typing.Iterable[str] | None = None, with_for_update: bool | None = None, auto_expunge: bool | None = None, auto_commit: bool | None = None, auto_refresh: bool | None = None, match_fields: list[str] | str | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.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] | ~typing.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 | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, typing.Any] | None = None) → ModelT

Wrap repository upsert operation.

Parameters:

• data – Instance to update existing, or be created. Identifier used to determine if an existing instance exists is the value of an attribute on data named as value of self.id_attribute.

• item_id – Identifier of the object for upsert.

• 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

• match_fields – a list of keys to use to match the existing model. When empty, all fields are matched.

• 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 or created representation.

upsert_many(data: BulkModelDictT[ModelT], *, auto_expunge: bool | None = None, auto_commit: bool | None = None, no_merge: bool = False, match_fields: list[str] | str | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) → Sequence[ModelT]

Wrap repository upsert operation.

Parameters:

• data – Instance to update existing, or be created. Identifier used to determine if an existing instance exists is the value of an attribute on data named as value of id_attribute.

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

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

• no_merge – Skip the usage of optimized Merge statements SQLAlchemyAsyncRepository.auto_commit

• match_fields – a list of keys to use to match the existing model. When empty, all fields are matched.

• 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 or created representation.

get_or_upsert(*filters: StatementFilter | ColumnElement[bool], match_fields: list[str] | str | None = None, upsert: bool = True, 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, **kwargs: Any) → tuple[ModelT, bool]

Wrap repository instance creation.

Parameters:

• *filters – Types for specific filtering operations.

• match_fields – a list of keys to use to match the existing model. When empty, all fields are matched.

• upsert – When using match_fields and actual model values differ from kwargs, perform an update operation on the model.

• create – Should a model be created. If no model is found, an exception is raised.

• 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

• 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

• **kwargs – Identifier of the instance to be retrieved.

Returns:

Representation of created instance.

get_and_update(*filters: StatementFilter | ColumnElement[bool], match_fields: list[str] | str | 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, **kwargs: Any) → tuple[ModelT, bool]

Wrap repository instance creation.

Parameters:

• *filters – Types for specific filtering operations.

• match_fields – a list of keys to use to match the existing model. When empty, all fields are matched.

• 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

• 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

• **kwargs – Identifier of the instance to be retrieved.

Returns:

Representation of updated instance.

delete(item_id: Any, *, auto_commit: bool | None = None, auto_expunge: bool | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) → ModelT

Wrap repository delete operation.

Parameters:

• item_id – Identifier of instance to be deleted.

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

• 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:

Representation of the deleted instance.

delete_many(item_ids: list[Any], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, chunk_size: int | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) → Sequence[ModelT]

Wrap repository bulk instance deletion.

Parameters:

• item_ids – Identifier of instance to be deleted.

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

• 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.

• chunk_size – Allows customization of the insertmanyvalues_max_parameters setting for the driver. Defaults to 950 if left unset.

• 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:

Representation of removed instances.

delete_where(*filters: StatementFilter | ColumnElement[bool], auto_commit: bool | None = None, auto_expunge: bool | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, sanity_check: bool = True, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None, **kwargs: Any) → Sequence[ModelT]

Wrap repository scalars operation.

Parameters:

• *filters – Types for specific filtering operations.

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

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

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

• sanity_check – When true, the length of selected instances is compared to the deleted row count

• load – Set default relationships to be loaded

• execution_options – Set default execution options

• **kwargs – Instance attribute value filters.

Returns:

The list of instances deleted from the repository.

class advanced_alchemy.service.SQLAlchemyAsyncRepositoryReadService

Bases: Generic[ModelT], ResultConverter

Service object that operates on a repository object.

__init__(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) → None

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 count(*filters: StatementFilter | ColumnElement[bool], statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None, **kwargs: Any) → int

Count of records returned by query.

Parameters:

• *filters – arguments for filtering.

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

• **kwargs – key value pairs of filter types.

Returns:

A count of the collection, filtered, but ignoring pagination.

async exists(*filters: StatementFilter | ColumnElement[bool], error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None, **kwargs: Any) → bool

Wrap repository exists operation.

Parameters:

• *filters – Types for specific filtering operations.

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

• **kwargs – Keyword arguments for attribute based filtering.

Returns:

Representation of instance with identifier item_id.

async get(item_id: Any, *, statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, auto_expunge: 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) → ModelT

Wrap repository scalar operation.

Parameters:

• item_id – Identifier of instance to be retrieved.

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

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

• 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 relationships to be loaded

• execution_options – Set default execution options

Returns:

Representation of instance with identifier item_id.

async get_one(*filters: StatementFilter | ColumnElement[bool], statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, auto_expunge: bool | None = None, load: LoadSpec | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, execution_options: dict[str, Any] | None = None, **kwargs: Any) → ModelT

Wrap repository scalar operation.

Parameters:

• *filters – Types for specific filtering operations.

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

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

• 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

• **kwargs – Identifier of the instance to be retrieved.

Returns:

Representation of instance with identifier item_id.

async get_one_or_none(*filters: StatementFilter | ColumnElement[bool], statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, auto_expunge: 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, **kwargs: Any) → ModelT | None

Wrap repository scalar operation.

Parameters:

• *filters – Types for specific filtering operations.

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

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

• 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

• **kwargs – Identifier of the instance to be retrieved.

Returns:

Representation of instance with identifier item_id.

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

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.

async list_and_count(*filters: StatementFilter | ColumnElement[bool], statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, auto_expunge: bool | None = None, force_basic_query_mode: bool | None = None, 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, **kwargs: Any) → tuple[Sequence[ModelT], int]

List of records and total count returned by query.

Parameters:

• *filters – Types for specific filtering operations.

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

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

• force_basic_query_mode – Force list and count to use two queries instead of an analytical window function.

• order_by – Set default order options for queries.

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

• **kwargs – Instance attribute value filters.

Returns:

List of instances and count of total collection, ignoring pagination.

classmethod new(session: AsyncSession | async_scoped_session[AsyncSession] | None = None, statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, config: SQLAlchemyAsyncConfig | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) → AsyncIterator[Self]

Context manager that returns instance of service object.

Handles construction of the database session._create_select_for_model

Returns:

The service object instance.

async list(*filters: StatementFilter | ColumnElement[bool], statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, auto_expunge: bool | None = None, 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, **kwargs: Any) → Sequence[ModelT]

Wrap repository scalars operation.

Parameters:

• *filters – Types for specific filtering operations.

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

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

• order_by – Set default order options for queries.

• 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

• **kwargs – Instance attribute value filters.

Returns:

The list of instances retrieved from the repository.

class advanced_alchemy.service.OffsetPagination

Bases: Generic[T]

Container for data returned using limit/offset pagination.

items: Sequence[T]

List of data being sent as part of the response.

limit: int

Maximal number of items to send.

__init__(items: Sequence[T], limit: int, offset: int, total: int) → None

offset: int

Offset from the beginning of the query.

Identical to an index.

total: int

Total number of items.

advanced_alchemy.service.find_filter(filter_type: type[FilterTypeT], filters: Sequence[StatementFilter | ColumnElement[bool]] | Sequence[StatementFilter]) → FilterTypeT | None

Get the filter specified by filter type from the filters.

Parameters:

• filter_type – The type of filter to find.

• filters – filter types to apply to the query

Returns:

The match filter instance or None

class advanced_alchemy.service.ResultConverter

Bases: object

Simple mixin to help convert to a paginated response model the results set is a list.

to_schema(data: ModelOrRowMappingT, total: int | None = None, filters: Sequence[StatementFilter | ColumnElement[bool]] | Sequence[StatementFilter] | None = None, *, schema_type: None = None) → ModelOrRowMappingT

to_schema(data: Sequence[ModelOrRowMappingT], total: int | None = None, filters: Sequence[StatementFilter | ColumnElement[bool]] | Sequence[StatementFilter] | None = None, *, schema_type: None = None) → OffsetPagination[ModelOrRowMappingT]

to_schema(data: ModelProtocol | RowMapping, total: int | None = None, filters: Sequence[StatementFilter | ColumnElement[bool]] | Sequence[StatementFilter] | None = None, *, schema_type: type[ModelDTOT]) → ModelDTOT

to_schema(data: Sequence[ModelProtocol] | Sequence[RowMapping], total: int | None = None, filters: Sequence[StatementFilter | ColumnElement[bool]] | Sequence[StatementFilter] | None = None, *, schema_type: type[ModelDTOT]) → OffsetPagination[ModelDTOT]

Convert the object to a response schema. When schema_type is None, the model is returned with no conversion.

Parameters:

• data – The return from one of the service calls.

• total – the total number of rows in the data

• filters – Collection route filters.

• schema_type – Collection route filters.

Returns:

The list of instances retrieved from the repository.

advanced_alchemy.service.model_from_dict(model: type[ModelT], **kwargs: Any) → ModelT

Return ORM Object from Dictionary.

final class advanced_alchemy.service.Empty

Bases: object

A sentinel class used as placeholder.

class advanced_alchemy.service.ErrorMessages

Bases: TypedDict