services

final class advanced_alchemy.service.Empty

Bases: object

A sentinel class used as placeholder.

class advanced_alchemy.service.ErrorMessages

Bases: TypedDict

duplicate_key: Union[str, Callable[[Exception], str]]

integrity: Union[str, Callable[[Exception], str]]

foreign_key: Union[str, Callable[[Exception], str]]

multiple_rows: Union[str, Callable[[Exception], str]]

check_constraint: Union[str, Callable[[Exception], str]]

other: Union[str, Callable[[Exception], str]]

class advanced_alchemy.service.OffsetPagination

Bases: Generic[T]

Container for data returned using limit/offset pagination.

items: Sequence[TypeVar(T)]

List of data being sent as part of the response.

limit: int

Maximal number of items to send.

__init__(items, limit, offset, total)

offset: int

Offset from the beginning of the query.

Identical to an index.

total: int

Total number of items.

class advanced_alchemy.service.ResultConverter

Bases: object

Simple mixin to help convert to a paginated response model.

Single objects are transformed to the supplied schema type, and lists of objects are automatically transformed into an OffsetPagination response of the supplied schema type.

Parameters:

data – A database model instance or row mapping. Type: ModelOrRowMappingT

Returns:

The converted schema object.

to_schema(data, total=None, filters=None, *, schema_type=None)

Convert the object to a response schema.

When schema_type is None, the model is returned with no conversion.

Parameters:

• data (Union[advanced_alchemy.repository.ModelOrRowMappingT, Sequence[advanced_alchemy.repository.ModelOrRowMappingT], ModelProtocol, Sequence[ModelProtocol], RowMapping, Sequence[RowMapping]]) – The return from one of the service calls. Type: ModelOrRowMappingT

• total (int | None) – The total number of rows in the data.

• filters (Union[Sequence[Union[StatementFilter, ColumnElement[bool]]], Sequence[StatementFilter], None]) – ColumnElement Collection of route filters.

• schema_type (type[advanced_alchemy.service.ModelDTOT] | None) – ModelDTOT Optional schema type to convert the data to

Return type:

Union[advanced_alchemy.repository.ModelOrRowMappingT, OffsetPagination[advanced_alchemy.repository.ModelOrRowMappingT], advanced_alchemy.service.ModelDTOT, OffsetPagination[advanced_alchemy.service.ModelDTOT]]

Returns:

• ModelProtocol | sqlalchemy.orm.RowMapping | OffsetPagination | msgspec.Struct | pydantic.BaseModel

class advanced_alchemy.service.SQLAlchemyAsyncQueryService

Bases: ResultConverter

Simple service to execute the basic Query repository..

__init__(session, **repo_kwargs)

Configure the service object.

Parameters:

• session (Union[AsyncSession, async_scoped_session[AsyncSession]]) – Session managing the unit-of-work for the operation.

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

classmethod new(cls, session=None, config=None)

Context manager that returns instance of service object.

Handles construction of the database session._create_select_for_model

Return type:

AsyncIterator[Self]

Returns:

The service object instance.

class advanced_alchemy.service.SQLAlchemyAsyncRepositoryReadService

Bases: ResultConverter, Generic[ModelT]

Service object that operates on a repository object.

repository_type: type[SQLAlchemyAsyncRepositoryProtocol[ModelT] | SQLAlchemyAsyncSlugRepositoryProtocol[ModelT]]

Type of the repository to use.

loader_options: LoadSpec | None = None

Default loader options for the repository.

execution_options: dict[str, Any] | None = None

Default execution options for the repository.

match_fields: list[str] | str | None = None

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

__init__(session, statement=None, auto_expunge=False, auto_refresh=True, auto_commit=False, order_by=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None, **repo_kwargs)

Configure the service object.

Parameters:

• session (AsyncSession | async_scoped_session[AsyncSession]) – Session managing the unit-of-work for the operation.

• statement (Select[tuple[ModelT]] | None) – To facilitate customization of the underlying select query.

• auto_expunge (bool) – Remove object from session before returning.

• auto_refresh (bool) – Refresh object from session before returning.

• auto_commit (bool) – Commit objects before returning.

• order_by (list[OrderingPair] | OrderingPair | None) – Set default order options for queries.

• error_messages (ErrorMessages | None | EmptyType) – A set of custom error messages to use for operations

• load (LoadSpec | None) – Set default relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

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

model_type: type[ModelT]

Type of the model to use.

async count(*filters, statement=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None, **kwargs)

Count of records returned by query.

Parameters:

• *filters (StatementFilter | ColumnElement[bool]) – arguments for filtering.

• statement (Select[tuple[ModelT]] | None) – To facilitate customization of the underlying select query.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

• **kwargs (Any) – key value pairs of filter types.

Return type:

int

Returns:

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

async exists(*filters, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None, **kwargs)

Wrap repository exists operation.

Parameters:

• *filters (StatementFilter | ColumnElement[bool]) – Types for specific filtering operations.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

• **kwargs (Any) – Keyword arguments for attribute based filtering.

Return type:

bool

Returns:

Representation of instance with identifier item_id.

async get(item_id, *, statement=None, id_attribute=None, auto_expunge=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None)

Wrap repository scalar operation.

Parameters:

• item_id – Identifier of instance to be retrieved.

• auto_expunge – Remove object from session before returning.

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

• 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, statement=None, auto_expunge=None, load=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, execution_options=None, **kwargs)

Wrap repository scalar operation.

Parameters:

• *filters (StatementFilter | ColumnElement[bool]) – Types for specific filtering operations.

• auto_expunge (bool | None) – Remove object from session before returning.

• statement (Select[tuple[ModelT]] | None) – To facilitate customization of the underlying select query.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set default relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

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

Return type:

ModelT

Returns:

Representation of instance with identifier item_id.

async get_one_or_none(*filters, statement=None, auto_expunge=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None, **kwargs)

Wrap repository scalar operation.

Parameters:

• *filters (StatementFilter | ColumnElement[bool]) – Types for specific filtering operations.

• auto_expunge (bool | None) – Remove object from session before returning.

• statement (Select[tuple[ModelT]] | None) – To facilitate customization of the underlying select query.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set default relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

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

Return type:

ModelT | None

Returns:

Representation of instance with identifier item_id.

async to_model(data, operation=None)

Parse and Convert input into a model.

Parameters:

• data (Union[Dict[str, Any], advanced_alchemy.repository.typing.ModelT, Struct, BaseModel, DTOData[advanced_alchemy.repository.typing.ModelT]]) – Representations to be created.

• operation (str | None) – Optional operation flag so that you can provide behavior based on CRUD operation

Return type:

advanced_alchemy.repository.typing.ModelT

Returns:

Representation of created instances.

async list_and_count(*filters, statement=None, auto_expunge=None, force_basic_query_mode=None, order_by=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None, **kwargs)

List of records and total count returned by query.

Parameters:

• *filters (StatementFilter | ColumnElement[bool]) – Types for specific filtering operations.

• statement (Select[tuple[ModelT]] | None) – To facilitate customization of the underlying select query.

• auto_expunge (bool | None) – Remove object from session before returning.

• force_basic_query_mode (bool | None) – Force list and count to use two queries instead of an analytical window function.

• order_by (list[OrderingPair] | OrderingPair | None) – Set default order options for queries.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

• **kwargs (Any) – Instance attribute value filters.

Return type:

tuple[Sequence[ModelT], int]

Returns:

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

classmethod new(cls, session=None, statement=None, config=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None)

Context manager that returns instance of service object.

Handles construction of the database session._create_select_for_model

Return type:

AsyncIterator[Self]

Returns:

The service object instance.

async list(*filters, statement=None, auto_expunge=None, order_by=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None, **kwargs)

Wrap repository scalars operation.

Parameters:

• *filters (StatementFilter | ColumnElement[bool]) – Types for specific filtering operations.

• auto_expunge (bool | None) – Remove object from session before returning.

• statement (Select[tuple[ModelT]] | None) – To facilitate customization of the underlying select query.

• order_by (list[OrderingPair] | OrderingPair | None) – Set default order options for queries.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set default relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

• **kwargs (Any) – Instance attribute value filters.

Return type:

Sequence[ModelT]

Returns:

The list of instances retrieved from the repository.

class advanced_alchemy.service.SQLAlchemyAsyncRepositoryService

Bases: SQLAlchemyAsyncRepositoryReadService[ModelT], Generic[ModelT]

Service object that operates on a repository object.

async create(data, *, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>)

Wrap repository instance creation.

Parameters:

• data (ModelDictT[ModelT]) – Representation to be created.

• auto_expunge (bool | None) – Remove object from session before returning.

• auto_refresh (bool | None) – Refresh object from session before returning.

• auto_commit (bool | None) – Commit objects before returning.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

Return type:

ModelT

Returns:

Representation of created instance.

async create_many(data, *, auto_commit=None, auto_expunge=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>)

Wrap repository bulk instance creation.

Parameters:

• data (BulkModelDictT[ModelT]) – Representations to be created.

• auto_expunge (bool | None) – Remove object from session before returning.

• auto_commit (bool | None) – Commit objects before returning.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

Return type:

Sequence[ModelT]

Returns:

Representation of created instances.

async update(data, item_id=None, *, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, id_attribute=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None)

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.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

• 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, *, auto_commit=None, auto_expunge=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None)

Wrap repository bulk instance update.

Parameters:

• data (BulkModelDictT[ModelT]) – Representations to be updated.

• auto_expunge (bool | None) – Remove object from session before returning.

• auto_commit (bool | None) – Commit objects before returning.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set default relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

Return type:

Sequence[ModelT]

Returns:

Representation of updated instances.

async upsert(data, item_id=None, *, attribute_names=None, with_for_update=None, auto_expunge=None, auto_commit=None, auto_refresh=None, match_fields=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None)

Wrap repository upsert operation.

Parameters:

• data (ModelDictT[ModelT]) – 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 (Any | None) – Identifier of the object for upsert.

• attribute_names (Iterable[str] | None) – an iterable of attribute names to pass into the update method.

• with_for_update (bool | None) – 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 (bool | None) – Remove object from session before returning.

• auto_refresh (bool | None) – Refresh object from session before returning.

• auto_commit (bool | None) – Commit objects before returning.

• match_fields (list[str] | str | None) – a list of keys to use to match the existing model. When empty, all fields are matched.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set default relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

Return type:

ModelT

Returns:

Updated or created representation.

async upsert_many(data, *, auto_expunge=None, auto_commit=None, no_merge=False, match_fields=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None)

Wrap repository upsert operation.

Parameters:

• data (BulkModelDictT[ModelT]) – Instance to update existing, or be created.

• auto_expunge (bool | None) – Remove object from session before returning.

• auto_commit (bool | None) – Commit objects before returning.

• no_merge (bool) – Skip the usage of optimized Merge statements (reserved for future use)

• match_fields (list[str] | str | None) – a list of keys to use to match the existing model. When empty, all fields are matched.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set default relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

Return type:

Sequence[ModelT]

Returns:

Updated or created representation.

async get_or_upsert(*filters, match_fields=None, upsert=True, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None, **kwargs)

Wrap repository instance creation.

Parameters:

• *filters (StatementFilter | ColumnElement[bool]) – Types for specific filtering operations.

• match_fields (list[str] | str | None) – a list of keys to use to match the existing model. When empty, all fields are matched.

• upsert (bool) – 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 (Iterable[str] | None) – an iterable of attribute names to pass into the update method.

• with_for_update (bool | None) – 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 (bool | None) – Remove object from session before returning.

• auto_refresh (bool | None) – Refresh object from session before returning.

• auto_commit (bool | None) – Commit objects before returning.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set default relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

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

Return type:

tuple[ModelT, bool]

Returns:

Representation of created instance.

async get_and_update(*filters, match_fields=None, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None, **kwargs)

Wrap repository instance creation.

Parameters:

• *filters (StatementFilter | ColumnElement[bool]) – Types for specific filtering operations.

• match_fields (list[str] | str | None) – a list of keys to use to match the existing model. When empty, all fields are matched.

• attribute_names (Iterable[str] | None) – an iterable of attribute names to pass into the update method.

• with_for_update (bool | None) – 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 (bool | None) – Remove object from session before returning.

• auto_refresh (bool | None) – Refresh object from session before returning.

• auto_commit (bool | None) – Commit objects before returning.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set default relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

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

Return type:

tuple[ModelT, bool]

Returns:

Representation of updated instance.

async delete(item_id, *, auto_commit=None, auto_expunge=None, id_attribute=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None)

Wrap repository delete operation.

Parameters:

• item_id – Identifier of instance to be deleted.

• auto_expunge – Remove object from session before returning.

• auto_commit – Commit objects before returning.

• 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, *, auto_commit=None, auto_expunge=None, id_attribute=None, chunk_size=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None)

Wrap repository bulk instance deletion.

Parameters:

• item_ids – Identifier of instance to be deleted.

• auto_expunge – Remove object from session before returning.

• auto_commit – Commit objects before returning.

• 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, auto_commit=None, auto_expunge=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, sanity_check=True, load=None, execution_options=None, **kwargs)

Wrap repository scalars operation.

Parameters:

• *filters (StatementFilter | ColumnElement[bool]) – Types for specific filtering operations.

• auto_expunge (bool | None) – Remove object from session before returning.

• auto_commit (bool | None) – Commit objects before returning.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

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

• load (LoadSpec | None) – Set default relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

• **kwargs (Any) – Instance attribute value filters.

Return type:

Sequence[ModelT]

Returns:

The list of instances deleted from the repository.

class advanced_alchemy.service.SQLAlchemySyncQueryService

Bases: ResultConverter

Simple service to execute the basic Query repository..

__init__(session, **repo_kwargs)

Configure the service object.

Parameters:

• session (Union[Session, scoped_session[Session]]) – Session managing the unit-of-work for the operation.

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

classmethod new(cls, session=None, config=None)

Context manager that returns instance of service object.

Handles construction of the database session._create_select_for_model

Return type:

Iterator[Self]

Returns:

The service object instance.

class advanced_alchemy.service.SQLAlchemySyncRepositoryReadService

Bases: ResultConverter, Generic[ModelT]

Service object that operates on a repository object.

repository_type: type[SQLAlchemySyncRepositoryProtocol[ModelT] | SQLAlchemySyncSlugRepositoryProtocol[ModelT]]

Type of the repository to use.

loader_options: LoadSpec | None = None

Default loader options for the repository.

execution_options: dict[str, Any] | None = None

Default execution options for the repository.

match_fields: list[str] | str | None = None

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

__init__(session, statement=None, auto_expunge=False, auto_refresh=True, auto_commit=False, order_by=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None, **repo_kwargs)

Configure the service object.

Parameters:

• session (Session | scoped_session[Session]) – Session managing the unit-of-work for the operation.

• statement (Select[tuple[ModelT]] | None) – To facilitate customization of the underlying select query.

• auto_expunge (bool) – Remove object from session before returning.

• auto_refresh (bool) – Refresh object from session before returning.

• auto_commit (bool) – Commit objects before returning.

• order_by (list[OrderingPair] | OrderingPair | None) – Set default order options for queries.

• error_messages (ErrorMessages | None | EmptyType) – A set of custom error messages to use for operations

• load (LoadSpec | None) – Set default relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

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

model_type: type[ModelT]

Type of the model to use.

count(*filters, statement=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None, **kwargs)

Count of records returned by query.

Parameters:

• *filters (StatementFilter | ColumnElement[bool]) – arguments for filtering.

• statement (Select[tuple[ModelT]] | None) – To facilitate customization of the underlying select query.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

• **kwargs (Any) – key value pairs of filter types.

Return type:

int

Returns:

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

exists(*filters, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None, **kwargs)

Wrap repository exists operation.

Parameters:

• *filters (StatementFilter | ColumnElement[bool]) – Types for specific filtering operations.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

• **kwargs (Any) – Keyword arguments for attribute based filtering.

Return type:

bool

Returns:

Representation of instance with identifier item_id.

get(item_id, *, statement=None, id_attribute=None, auto_expunge=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None)

Wrap repository scalar operation.

Parameters:

• item_id – Identifier of instance to be retrieved.

• auto_expunge – Remove object from session before returning.

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

• 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, statement=None, auto_expunge=None, load=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, execution_options=None, **kwargs)

Wrap repository scalar operation.

Parameters:

• *filters (StatementFilter | ColumnElement[bool]) – Types for specific filtering operations.

• auto_expunge (bool | None) – Remove object from session before returning.

• statement (Select[tuple[ModelT]] | None) – To facilitate customization of the underlying select query.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set default relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

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

Return type:

ModelT

Returns:

Representation of instance with identifier item_id.

get_one_or_none(*filters, statement=None, auto_expunge=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None, **kwargs)

Wrap repository scalar operation.

Parameters:

• *filters (StatementFilter | ColumnElement[bool]) – Types for specific filtering operations.

• auto_expunge (bool | None) – Remove object from session before returning.

• statement (Select[tuple[ModelT]] | None) – To facilitate customization of the underlying select query.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set default relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

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

Return type:

ModelT | None

Returns:

Representation of instance with identifier item_id.

to_model(data, operation=None)

Parse and Convert input into a model.

Parameters:

• data (Union[Dict[str, Any], advanced_alchemy.repository.typing.ModelT, Struct, BaseModel, DTOData[advanced_alchemy.repository.typing.ModelT]]) – Representations to be created.

• operation (str | None) – Optional operation flag so that you can provide behavior based on CRUD operation

Return type:

advanced_alchemy.repository.typing.ModelT

Returns:

Representation of created instances.

list_and_count(*filters, statement=None, auto_expunge=None, force_basic_query_mode=None, order_by=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None, **kwargs)

List of records and total count returned by query.

Parameters:

• *filters (StatementFilter | ColumnElement[bool]) – Types for specific filtering operations.

• statement (Select[tuple[ModelT]] | None) – To facilitate customization of the underlying select query.

• auto_expunge (bool | None) – Remove object from session before returning.

• force_basic_query_mode (bool | None) – Force list and count to use two queries instead of an analytical window function.

• order_by (list[OrderingPair] | OrderingPair | None) – Set default order options for queries.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

• **kwargs (Any) – Instance attribute value filters.

Return type:

tuple[Sequence[ModelT], int]

Returns:

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

classmethod new(cls, session=None, statement=None, config=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None)

Context manager that returns instance of service object.

Handles construction of the database session._create_select_for_model

Return type:

Iterator[Self]

Returns:

The service object instance.

list(*filters, statement=None, auto_expunge=None, order_by=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None, **kwargs)

Wrap repository scalars operation.

Parameters:

• *filters (StatementFilter | ColumnElement[bool]) – Types for specific filtering operations.

• auto_expunge (bool | None) – Remove object from session before returning.

• statement (Select[tuple[ModelT]] | None) – To facilitate customization of the underlying select query.

• order_by (list[OrderingPair] | OrderingPair | None) – Set default order options for queries.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set default relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

• **kwargs (Any) – Instance attribute value filters.

Return type:

Sequence[ModelT]

Returns:

The list of instances retrieved from the repository.

class advanced_alchemy.service.SQLAlchemySyncRepositoryService

Bases: SQLAlchemySyncRepositoryReadService[ModelT], Generic[ModelT]

Service object that operates on a repository object.

create(data, *, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>)

Wrap repository instance creation.

Parameters:

• data (ModelDictT[ModelT]) – Representation to be created.

• auto_expunge (bool | None) – Remove object from session before returning.

• auto_refresh (bool | None) – Refresh object from session before returning.

• auto_commit (bool | None) – Commit objects before returning.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

Return type:

ModelT

Returns:

Representation of created instance.

create_many(data, *, auto_commit=None, auto_expunge=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>)

Wrap repository bulk instance creation.

Parameters:

• data (BulkModelDictT[ModelT]) – Representations to be created.

• auto_expunge (bool | None) – Remove object from session before returning.

• auto_commit (bool | None) – Commit objects before returning.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

Return type:

Sequence[ModelT]

Returns:

Representation of created instances.

update(data, item_id=None, *, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, id_attribute=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None)

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.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

• 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, *, auto_commit=None, auto_expunge=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None)

Wrap repository bulk instance update.

Parameters:

• data (BulkModelDictT[ModelT]) – Representations to be updated.

• auto_expunge (bool | None) – Remove object from session before returning.

• auto_commit (bool | None) – Commit objects before returning.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set default relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

Return type:

Sequence[ModelT]

Returns:

Representation of updated instances.

upsert(data, item_id=None, *, attribute_names=None, with_for_update=None, auto_expunge=None, auto_commit=None, auto_refresh=None, match_fields=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None)

Wrap repository upsert operation.

Parameters:

• data (ModelDictT[ModelT]) – 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 (Any | None) – Identifier of the object for upsert.

• attribute_names (Iterable[str] | None) – an iterable of attribute names to pass into the update method.

• with_for_update (bool | None) – 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 (bool | None) – Remove object from session before returning.

• auto_refresh (bool | None) – Refresh object from session before returning.

• auto_commit (bool | None) – Commit objects before returning.

• match_fields (list[str] | str | None) – a list of keys to use to match the existing model. When empty, all fields are matched.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set default relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

Return type:

ModelT

Returns:

Updated or created representation.

upsert_many(data, *, auto_expunge=None, auto_commit=None, no_merge=False, match_fields=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None)

Wrap repository upsert operation.

Parameters:

• data (BulkModelDictT[ModelT]) – Instance to update existing, or be created.

• auto_expunge (bool | None) – Remove object from session before returning.

• auto_commit (bool | None) – Commit objects before returning.

• no_merge (bool) – Skip the usage of optimized Merge statements (reserved for future use)

• match_fields (list[str] | str | None) – a list of keys to use to match the existing model. When empty, all fields are matched.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set default relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

Return type:

Sequence[ModelT]

Returns:

Updated or created representation.

get_or_upsert(*filters, match_fields=None, upsert=True, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None, **kwargs)

Wrap repository instance creation.

Parameters:

• *filters (StatementFilter | ColumnElement[bool]) – Types for specific filtering operations.

• match_fields (list[str] | str | None) – a list of keys to use to match the existing model. When empty, all fields are matched.

• upsert (bool) – 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 (Iterable[str] | None) – an iterable of attribute names to pass into the update method.

• with_for_update (bool | None) – 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 (bool | None) – Remove object from session before returning.

• auto_refresh (bool | None) – Refresh object from session before returning.

• auto_commit (bool | None) – Commit objects before returning.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set default relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

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

Return type:

tuple[ModelT, bool]

Returns:

Representation of created instance.

get_and_update(*filters, match_fields=None, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None, **kwargs)

Wrap repository instance creation.

Parameters:

• *filters (StatementFilter | ColumnElement[bool]) – Types for specific filtering operations.

• match_fields (list[str] | str | None) – a list of keys to use to match the existing model. When empty, all fields are matched.

• attribute_names (Iterable[str] | None) – an iterable of attribute names to pass into the update method.

• with_for_update (bool | None) – 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 (bool | None) – Remove object from session before returning.

• auto_refresh (bool | None) – Refresh object from session before returning.

• auto_commit (bool | None) – Commit objects before returning.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

• load (LoadSpec | None) – Set default relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

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

Return type:

tuple[ModelT, bool]

Returns:

Representation of updated instance.

delete(item_id, *, auto_commit=None, auto_expunge=None, id_attribute=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None)

Wrap repository delete operation.

Parameters:

• item_id – Identifier of instance to be deleted.

• auto_expunge – Remove object from session before returning.

• auto_commit – Commit objects before returning.

• 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, *, auto_commit=None, auto_expunge=None, id_attribute=None, chunk_size=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None)

Wrap repository bulk instance deletion.

Parameters:

• item_ids – Identifier of instance to be deleted.

• auto_expunge – Remove object from session before returning.

• auto_commit – Commit objects before returning.

• 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, auto_commit=None, auto_expunge=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, sanity_check=True, load=None, execution_options=None, **kwargs)

Wrap repository scalars operation.

Parameters:

• *filters (StatementFilter | ColumnElement[bool]) – Types for specific filtering operations.

• auto_expunge (bool | None) – Remove object from session before returning.

• auto_commit (bool | None) – Commit objects before returning.

• error_messages (ErrorMessages | None | EmptyType) – An optional dictionary of templates to use for friendlier error messages to clients

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

• load (LoadSpec | None) – Set default relationships to be loaded

• execution_options (dict[str, Any] | None) – Set default execution options

• **kwargs (Any) – Instance attribute value filters.

Return type:

Sequence[ModelT]

Returns:

The list of instances deleted from the repository.

advanced_alchemy.service.find_filter(filter_type, filters)

Get the filter specified by filter type from the filters.

Parameters:

• filter_type (type[FilterTypeT]) – The type of filter to find.

• filters (Sequence[StatementFilter | ColumnElement[bool]] | Sequence[StatementFilter]) – filter types to apply to the query

Return type:

FilterTypeT | None

Returns:

The match filter instance or None

advanced_alchemy.service.is_dict(v)

Check if a value is a dictionary.

Parameters:

v (Any) – Value to check.

Return type:

TypeGuard[dict[str, Any]]

Returns:

bool

advanced_alchemy.service.is_dict_with_field(v, field_name)

Check if a dictionary has a specific field.

Parameters:

• v (Any) – Value to check.

• field_name (str) – Field name to check for.

Return type:

TypeGuard[dict[str, Any]]

Returns:

bool

advanced_alchemy.service.is_dict_without_field(v, field_name)

Check if a dictionary does not have a specific field.

Parameters:

• v (Any) – Value to check.

• field_name (str) – Field name to check for.

Return type:

TypeGuard[dict[str, Any]]

Returns:

bool

advanced_alchemy.service.is_dto_data(v)

Check if a value is a Litestar DTOData object.

Parameters:

v – Value to check.

Returns:

bool

advanced_alchemy.service.is_msgspec_struct(v)

Check if a value is a msgspec struct.

Parameters:

v (Any) – Value to check.

Return type:

TypeGuard[msgspec.Struct]

Returns:

bool

advanced_alchemy.service.is_msgspec_struct_with_field(v, field_name)

Check if a msgspec struct has a specific field.

Parameters:

• v (Any) – Value to check.

• field_name (str) – Field name to check for.

Return type:

TypeGuard[msgspec.Struct]

Returns:

bool

advanced_alchemy.service.is_msgspec_struct_without_field(v, field_name)

Check if a msgspec struct does not have a specific field.

Parameters:

• v (Any) – Value to check.

• field_name (str) – Field name to check for.

Return type:

TypeGuard[msgspec.Struct]

Returns:

bool

advanced_alchemy.service.is_pydantic_model(v)

Check if a value is a pydantic model.

Parameters:

v (Any) – Value to check.

Return type:

TypeGuard[pydantic.BaseModel]

Returns:

bool

advanced_alchemy.service.is_pydantic_model_with_field(v, field_name)

Check if a pydantic model has a specific field.

Parameters:

• v (Any) – Value to check.

• field_name (str) – Field name to check for.

Return type:

TypeGuard[pydantic.BaseModel]

Returns:

bool

advanced_alchemy.service.is_pydantic_model_without_field(v, field_name)

Check if a pydantic model does not have a specific field.

Parameters:

• v (Any) – Value to check.

• field_name (str) – Field name to check for.

Return type:

TypeGuard[pydantic.BaseModel]

Returns:

bool

advanced_alchemy.service.is_schema(v)

Check if a value is a msgspec Struct or Pydantic model.

Parameters:

v (Any) – Value to check.

Return type:

TypeGuard[Struct | BaseModel]

Returns:

bool

advanced_alchemy.service.is_schema_or_dict(v)

Check if a value is a msgspec Struct, Pydantic model, or dict.

Parameters:

v (Any) – Value to check.

Return type:

TypeGuard[Struct | BaseModel | dict[str, Any]]

Returns:

bool

advanced_alchemy.service.is_schema_or_dict_with_field(v, field_name)

Check if a value is a msgspec Struct, Pydantic model, or dict with a specific field.

Parameters:

• v (Any) – Value to check.

• field_name (str) – Field name to check for.

Return type:

TypeGuard[Struct | BaseModel | dict[str, Any]]

Returns:

bool

advanced_alchemy.service.is_schema_or_dict_without_field(v, field_name)

Check if a value is a msgspec Struct, Pydantic model, or dict without a specific field.

Parameters:

• v (Any) – Value to check.

• field_name (str) – Field name to check for.

Return type:

TypeGuard[Struct | BaseModel | dict[str, Any]]

Returns:

bool

advanced_alchemy.service.is_schema_with_field(v, field_name)

Check if a value is a msgspec Struct or Pydantic model with a specific field.

Parameters:

• v (Any) – Value to check.

• field_name (str) – Field name to check for.

Return type:

TypeGuard[Struct | BaseModel]

Returns:

bool

advanced_alchemy.service.is_schema_without_field(v, field_name)

Check if a value is a msgspec Struct or Pydantic model without a specific field.

Parameters:

• v (Any) – Value to check.

• field_name (str) – Field name to check for.

Return type:

TypeGuard[Struct | BaseModel]

Returns:

bool

advanced_alchemy.service.model_from_dict(model, **kwargs)

Create an ORM model instance from a dictionary of attributes.

Parameters:

• model (type[advanced_alchemy.repository.typing.ModelT]) – The SQLAlchemy model class to instantiate.

• **kwargs (Any) – Keyword arguments containing model attribute values.

Returns:

A new instance of the model populated with the provided values.

Return type:

ModelT

advanced_alchemy.service.schema_dump(data, exclude_unset=True)

Dump a data object to a dictionary.

Parameters:

• data (dict[str, Any] | ModelT | Struct | BaseModel | DTOData[ModelT]) – dict[str, Any] | advanced_alchemy.base.ModelProtocol | msgspec.Struct | pydantic.BaseModel | litestar.dto.data_structures.DTOData[ModelT]

• exclude_unset (bool) – bool Whether to exclude unset values.

Returns:

type: dict[str, Any], ModelProtocol]

Return type:

Union[