services

class advanced_alchemy.service.AttrsInstance

Bases: Protocol

__init__(*args, **kwargs)

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

not_found: 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.

offset: int

Offset from the beginning of the query.

Identical to an index.

total: int

Total number of items.

__init__(items, limit, offset, total)

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 – The return from one of the service calls. Type: ModelOrRowMappingT

• total – The total number of rows in the data.

• filters – ColumnElement Collection of route filters.

• schema_type – ModelDTOT Optional schema type to convert the data to

Raises:

AdvancedAlchemyError – If schema_type is not a valid Pydantic, Msgspec, or attrs schema and all libraries are not installed.

Returns:

ModelProtocol | sqlalchemy.orm.RowMapping | OffsetPagination | msgspec.Struct | pydantic.BaseModel | attrs class

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

Raises:

AdvancedAlchemyError – If no configuration or session is provided.

Yields:

The service object instance.

Return type:

AsyncIterator[Self]

class advanced_alchemy.service.SQLAlchemyAsyncRepositoryReadService

Bases: ResultConverter, Generic[ModelT, SQLAlchemyAsyncRepositoryT]

Service object that operates on a repository object.

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

Configure the service object.

Parameters:

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

• statement (Optional[Select]) – 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 (Union[list[Union[tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any]]], tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any], None]) – Set default order options for queries.

• error_messages (Union[ErrorMessages, type[Empty], None]) – A set of custom error messages to use for operations

• wrap_exceptions (bool) – Wrap exceptions in a RepositoryError

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

• count_with_window_function (Optional[bool]) – When false, list and count will use two queries instead of an analytical window function.

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

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

Count of records returned by query.

Parameters:

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

• statement (Optional[Select]) – To facilitate customization of the underlying select query.

• error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

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

Return type:

int

Returns:

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

count_with_window_function: ClassVar[bool] = True

Use an analytical window function to count results. This allows the count to be performed in a single query.

execution_options: ClassVar[Optional[dict[str, Any]]] = None

Default execution options for the repository.

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

Wrap repository exists operation.

Parameters:

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

• error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

• **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, uniquify=None)

Wrap repository scalar operation.

Parameters:

• item_id (Any) – Identifier of instance to be retrieved.

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

• statement (Optional[Select]) – To facilitate customization of the underlying select query.

• id_attribute (Union[str, InstrumentedAttribute[Any], None]) – 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 (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

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, uniquify=None, **kwargs)

Wrap repository scalar operation.

Parameters:

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

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

• statement (Optional[Select]) – To facilitate customization of the underlying select query.

• error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

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

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

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, uniquify=None, **kwargs)

Wrap repository scalar operation.

Parameters:

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

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

• statement (Optional[Select]) – To facilitate customization of the underlying select query.

• error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

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

Return type:

Optional[TypeVar(ModelT, bound= base.ModelProtocol)]

Returns:

Representation of instance with identifier item_id.

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

Wrap repository scalars operation.

Parameters:

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

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

• statement (Optional[Select]) – To facilitate customization of the underlying select query.

• order_by (Union[list[Union[tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any]]], tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any], None]) – Set default order options for queries.

• error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

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

Return type:

Sequence[TypeVar(ModelT, bound= base.ModelProtocol)]

Returns:

The list of instances retrieved from the repository.

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

List of records and total count returned by query.

Parameters:

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

• statement (Optional[Select]) – To facilitate customization of the underlying select query.

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

• count_with_window_function (Optional[bool]) – When false, list and count will use two queries instead of an analytical window function.

• order_by (Union[list[Union[tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any]]], tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any], None]) – Set default order options for queries.

• error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

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

Return type:

tuple[Sequence[TypeVar(ModelT, bound= base.ModelProtocol)], int]

Returns:

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

loader_options: ClassVar[Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]] = None

Default loader options for the repository.

match_fields: ClassVar[Union[list[str], str, None]] = None

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

property model_type: type[ModelT]

Return the model type.

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

Context manager that returns instance of service object.

Handles construction of the database session._create_select_for_model

Raises:

AdvancedAlchemyError – If no configuration or session is provided.

Yields:

The service object instance.

Return type:

AsyncIterator[Self]

property repository: SQLAlchemyAsyncRepositoryT

Return the repository instance.

Raises:

ImproperConfigurationError – If the repository is not initialized.

Returns:

The repository instance.

async to_model(data, operation=None)

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 to_model_on_create(data)

Convenience method to allow for custom behavior on create.

Parameters:

data – The data to be converted to a model.

Returns:

The data to be converted to a model.

async to_model_on_delete(data)

Convenience method to allow for custom behavior on delete.

Parameters:

data – The data to be converted to a model.

Returns:

The data to be converted to a model.

async to_model_on_update(data)

Convenience method to allow for custom behavior on update.

Parameters:

data – The data to be converted to a model.

Returns:

The data to be converted to a model.

async to_model_on_upsert(data)

Convenience method to allow for custom behavior on upsert.

Parameters:

data – The data to be converted to a model.

Returns:

The data to be converted to a model.

uniquify: ClassVar[bool] = False

Optionally apply the unique() method to results before returning.

repository_type: type[TypeVar(SQLAlchemyAsyncRepositoryT, bound= Union[SQLAlchemyAsyncRepository[Any], SQLAlchemyAsyncMockRepository[Any]])]

Type of the repository to use.

class advanced_alchemy.service.SQLAlchemyAsyncRepositoryService

Bases: SQLAlchemyAsyncRepositoryReadService[ModelT, SQLAlchemyAsyncRepositoryT], Generic[ModelT, SQLAlchemyAsyncRepositoryT]

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 – Representation to be created.

• auto_expunge – Remove object from session before returning.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

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

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 – Representations to be created.

• auto_expunge – Remove object from session before returning.

• auto_commit – Commit objects before returning.

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

Returns:

Representation of created instances.

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, uniquify=None)

Wrap repository delete operation.

Parameters:

• item_id (Any) – Identifier of instance to be deleted.

• auto_commit (Optional[bool]) – Commit objects before returning.

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

• id_attribute (Union[str, InstrumentedAttribute[Any], None]) – 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 (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

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, uniquify=None)

Wrap repository bulk instance deletion.

Parameters:

• item_ids (list[Any]) – Identifier of instance to be deleted.

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

• auto_commit (Optional[bool]) – Commit objects before returning.

• id_attribute (Union[str, InstrumentedAttribute[Any], None]) – 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 (Optional[int]) – Allows customization of the insertmanyvalues_max_parameters setting for the driver. Defaults to 950 if left unset.

• error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

Return type:

Sequence[TypeVar(ModelT, bound= base.ModelProtocol)]

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, uniquify=None, **kwargs)

Wrap repository scalars operation.

Parameters:

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

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

• auto_commit (Optional[bool]) – Commit objects before returning.

• error_messages (Union[ErrorMessages, type[Empty], None]) – 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 (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

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

Return type:

Sequence[TypeVar(ModelT, bound= base.ModelProtocol)]

Returns:

The list of instances deleted from the repository.

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, uniquify=None, **kwargs)

Wrap repository instance creation.

Parameters:

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

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

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

• with_for_update (Union[sqlalchemy.sql.selectable.ForUpdateArg, None, bool, Dict[str, Any]]) – 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 (Optional[bool]) – Remove object from session before returning.

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

• auto_commit (Optional[bool]) – Commit objects before returning.

• error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

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

Return type:

tuple[TypeVar(ModelT, bound= base.ModelProtocol), bool]

Returns:

Representation of updated instance.

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, uniquify=None, **kwargs)

Wrap repository instance creation.

Parameters:

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

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

• with_for_update (Union[sqlalchemy.sql.selectable.ForUpdateArg, None, bool, Dict[str, Any]]) – 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 (Optional[bool]) – Remove object from session before returning.

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

• auto_commit (Optional[bool]) – Commit objects before returning.

• error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

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

Return type:

tuple[TypeVar(ModelT, bound= base.ModelProtocol), bool]

Returns:

Representation of created instance.

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, uniquify=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

• uniquify – Optionally apply the unique() method to results before returning.

Raises:

RepositoryError – If no configuration or session is provided.

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, uniquify=None)

Wrap repository bulk instance update.

Parameters:

• data – Representations to be updated.

• auto_expunge – Remove object from session before returning.

• auto_commit – Commit objects before returning.

• 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

• uniquify – Optionally apply the unique() method to results before returning.

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, uniquify=None)

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.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

• 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

• uniquify – Optionally apply the unique() method to results before returning.

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, uniquify=None)

Wrap repository upsert operation.

Parameters:

• data – Instance to update existing, or be created.

• auto_expunge – Remove object from session before returning.

• auto_commit – Commit objects before returning.

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

• 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

• uniquify – Optionally apply the unique() method to results before returning.

Returns:

Updated or created representation.

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

Raises:

AdvancedAlchemyError – If no configuration or session is provided.

Yields:

The service object instance.

Return type:

Iterator[Self]

class advanced_alchemy.service.SQLAlchemySyncRepositoryReadService

Bases: ResultConverter, Generic[ModelT, SQLAlchemySyncRepositoryT]

Service object that operates on a repository object.

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

Configure the service object.

Parameters:

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

• statement (Optional[Select]) – 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 (Union[list[Union[tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any]]], tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any], None]) – Set default order options for queries.

• error_messages (Union[ErrorMessages, type[Empty], None]) – A set of custom error messages to use for operations

• wrap_exceptions (bool) – Wrap exceptions in a RepositoryError

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

• count_with_window_function (Optional[bool]) – When false, list and count will use two queries instead of an analytical window function.

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

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

Count of records returned by query.

Parameters:

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

• statement (Optional[Select]) – To facilitate customization of the underlying select query.

• error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

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

Return type:

int

Returns:

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

count_with_window_function: ClassVar[bool] = True

Use an analytical window function to count results. This allows the count to be performed in a single query.

execution_options: ClassVar[Optional[dict[str, Any]]] = None

Default execution options for the repository.

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

Wrap repository exists operation.

Parameters:

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

• error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

• **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, uniquify=None)

Wrap repository scalar operation.

Parameters:

• item_id (Any) – Identifier of instance to be retrieved.

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

• statement (Optional[Select]) – To facilitate customization of the underlying select query.

• id_attribute (Union[str, InstrumentedAttribute[Any], None]) – 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 (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

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, uniquify=None, **kwargs)

Wrap repository scalar operation.

Parameters:

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

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

• statement (Optional[Select]) – To facilitate customization of the underlying select query.

• error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

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

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

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, uniquify=None, **kwargs)

Wrap repository scalar operation.

Parameters:

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

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

• statement (Optional[Select]) – To facilitate customization of the underlying select query.

• error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

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

Return type:

Optional[TypeVar(ModelT, bound= base.ModelProtocol)]

Returns:

Representation of instance with identifier item_id.

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

Wrap repository scalars operation.

Parameters:

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

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

• statement (Optional[Select]) – To facilitate customization of the underlying select query.

• order_by (Union[list[Union[tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any]]], tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any], None]) – Set default order options for queries.

• error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

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

Return type:

Sequence[TypeVar(ModelT, bound= base.ModelProtocol)]

Returns:

The list of instances retrieved from the repository.

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

List of records and total count returned by query.

Parameters:

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

• statement (Optional[Select]) – To facilitate customization of the underlying select query.

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

• count_with_window_function (Optional[bool]) – When false, list and count will use two queries instead of an analytical window function.

• order_by (Union[list[Union[tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any]]], tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any], None]) – Set default order options for queries.

• error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

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

Return type:

tuple[Sequence[TypeVar(ModelT, bound= base.ModelProtocol)], int]

Returns:

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

loader_options: ClassVar[Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]] = None

Default loader options for the repository.

match_fields: ClassVar[Union[list[str], str, None]] = None

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

property model_type: type[ModelT]

Return the model type.

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

Context manager that returns instance of service object.

Handles construction of the database session._create_select_for_model

Raises:

AdvancedAlchemyError – If no configuration or session is provided.

Yields:

The service object instance.

Return type:

Iterator[Self]

property repository: SQLAlchemySyncRepositoryT

Return the repository instance.

Raises:

ImproperConfigurationError – If the repository is not initialized.

Returns:

The repository instance.

to_model(data, operation=None)

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.

to_model_on_create(data)

Convenience method to allow for custom behavior on create.

Parameters:

data – The data to be converted to a model.

Returns:

The data to be converted to a model.

to_model_on_delete(data)

Convenience method to allow for custom behavior on delete.

Parameters:

data – The data to be converted to a model.

Returns:

The data to be converted to a model.

to_model_on_update(data)

Convenience method to allow for custom behavior on update.

Parameters:

data – The data to be converted to a model.

Returns:

The data to be converted to a model.

to_model_on_upsert(data)

Convenience method to allow for custom behavior on upsert.

Parameters:

data – The data to be converted to a model.

Returns:

The data to be converted to a model.

uniquify: ClassVar[bool] = False

Optionally apply the unique() method to results before returning.

repository_type: type[TypeVar(SQLAlchemySyncRepositoryT, bound= Union[SQLAlchemySyncRepository[Any], SQLAlchemySyncMockRepository[Any]])]

Type of the repository to use.

class advanced_alchemy.service.SQLAlchemySyncRepositoryService

Bases: SQLAlchemySyncRepositoryReadService[ModelT, SQLAlchemySyncRepositoryT], Generic[ModelT, SQLAlchemySyncRepositoryT]

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 – Representation to be created.

• auto_expunge – Remove object from session before returning.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

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

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 – Representations to be created.

• auto_expunge – Remove object from session before returning.

• auto_commit – Commit objects before returning.

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

Returns:

Representation of created instances.

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, uniquify=None)

Wrap repository delete operation.

Parameters:

• item_id (Any) – Identifier of instance to be deleted.

• auto_commit (Optional[bool]) – Commit objects before returning.

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

• id_attribute (Union[str, InstrumentedAttribute[Any], None]) – 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 (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

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, uniquify=None)

Wrap repository bulk instance deletion.

Parameters:

• item_ids (list[Any]) – Identifier of instance to be deleted.

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

• auto_commit (Optional[bool]) – Commit objects before returning.

• id_attribute (Union[str, InstrumentedAttribute[Any], None]) – 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 (Optional[int]) – Allows customization of the insertmanyvalues_max_parameters setting for the driver. Defaults to 950 if left unset.

• error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

Return type:

Sequence[TypeVar(ModelT, bound= base.ModelProtocol)]

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, uniquify=None, **kwargs)

Wrap repository scalars operation.

Parameters:

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

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

• auto_commit (Optional[bool]) – Commit objects before returning.

• error_messages (Union[ErrorMessages, type[Empty], None]) – 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 (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

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

Return type:

Sequence[TypeVar(ModelT, bound= base.ModelProtocol)]

Returns:

The list of instances deleted from the repository.

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, uniquify=None, **kwargs)

Wrap repository instance creation.

Parameters:

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

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

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

• with_for_update (Union[sqlalchemy.sql.selectable.ForUpdateArg, None, bool, Dict[str, Any]]) – 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 (Optional[bool]) – Remove object from session before returning.

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

• auto_commit (Optional[bool]) – Commit objects before returning.

• error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

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

Return type:

tuple[TypeVar(ModelT, bound= base.ModelProtocol), bool]

Returns:

Representation of updated instance.

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, uniquify=None, **kwargs)

Wrap repository instance creation.

Parameters:

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

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

• with_for_update (Union[sqlalchemy.sql.selectable.ForUpdateArg, None, bool, Dict[str, Any]]) – 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 (Optional[bool]) – Remove object from session before returning.

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

• auto_commit (Optional[bool]) – Commit objects before returning.

• error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

• load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

• uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

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

Return type:

tuple[TypeVar(ModelT, bound= base.ModelProtocol), bool]

Returns:

Representation of created instance.

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, uniquify=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

• uniquify – Optionally apply the unique() method to results before returning.

Raises:

RepositoryError – If no configuration or session is provided.

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, uniquify=None)

Wrap repository bulk instance update.

Parameters:

• data – Representations to be updated.

• auto_expunge – Remove object from session before returning.

• auto_commit – Commit objects before returning.

• 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

• uniquify – Optionally apply the unique() method to results before returning.

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, uniquify=None)

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.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

• 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

• uniquify – Optionally apply the unique() method to results before returning.

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, uniquify=None)

Wrap repository upsert operation.

Parameters:

• data – Instance to update existing, or be created.

• auto_expunge – Remove object from session before returning.

• auto_commit – Commit objects before returning.

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

• 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

• uniquify – Optionally apply the unique() method to results before returning.

Returns:

Updated or created representation.

advanced_alchemy.service.fields(cls)

Return the tuple of attrs attributes for a class.

The tuple also allows accessing the fields by their names (see below for examples).

Parameters:

cls (type) – Class to introspect.

Raises:

• TypeError – If cls is not a class.

• attrs.exceptions.NotAnAttrsClassError – If cls is not an attrs class.

Returns:

tuple (with name accessors) of attrs.Attribute

Changed in version 16.2.0: Returned tuple allows accessing the fields by name.

Changed in version 23.1.0: Add support for generic classes.

advanced_alchemy.service.find_filter(filter_type, filters)

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

advanced_alchemy.service.is_attrs_instance(obj)

Check if a value is an attrs class instance.

Parameters:

obj (Any) – Value to check.

Return type:

TypeGuard[AttrsLike]

Returns:

bool

advanced_alchemy.service.is_attrs_instance_with_field(v, field_name)

Check if an attrs instance has a specific field.

Parameters:

• v (Any) – Value to check.

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

Return type:

TypeGuard[AttrsLike]

Returns:

bool

advanced_alchemy.service.is_attrs_instance_without_field(v, field_name)

Check if an attrs instance does not have a specific field.

Parameters:

• v (Any) – Value to check.

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

Return type:

TypeGuard[AttrsLike]

Returns:

bool

advanced_alchemy.service.is_attrs_schema(cls)

Check if a class type is an attrs schema.

Parameters:

cls (Any) – Class to check.

Return type:

TypeGuard[type[AttrsLike]]

Returns:

bool

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 (Any) – Value to check.

Return type:

TypeGuard[DTODataLike[Any]]

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[StructLike]

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[StructLike]

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[StructLike]

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[BaseModelLike]

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[BaseModelLike]

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[BaseModelLike]

Returns:

bool

advanced_alchemy.service.is_schema(v)

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

Parameters:

v (Any) – Value to check.

Return type:

TypeGuard[Union[StructLike, BaseModelLike, AttrsLike]]

Returns:

bool

advanced_alchemy.service.is_schema_or_dict(v)

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

Parameters:

v – Value to check.

Returns:

bool

advanced_alchemy.service.is_schema_or_dict_with_field(v, field_name)

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

Parameters:

• v – Value to check.

• field_name – Field name to check for.

Returns:

bool

advanced_alchemy.service.is_schema_or_dict_without_field(v, field_name)

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

Parameters:

• v – Value to check.

• field_name – Field name to check for.

Returns:

bool

advanced_alchemy.service.is_schema_with_field(v, field_name)

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

Parameters:

• v (Any) – Value to check.

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

Return type:

TypeGuard[Union[StructLike, BaseModelLike, AttrsLike]]

Returns:

bool

advanced_alchemy.service.is_schema_without_field(v, field_name)

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

Parameters:

• v (Any) – Value to check.

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

Return type:

TypeGuard[Union[StructLike, BaseModelLike, AttrsLike]]

Returns:

bool

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

Create an ORM model instance from a dictionary of attributes.

Parameters:

• model (type[TypeVar(ModelT, bound= base.ModelProtocol)]) – 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] | advanced_alchemy.base.ModelProtocol | msgspec.Struct | pydantic.BaseModel | attrs class | litestar.dto.data_structures.DTOData[ModelT] | sqlalchemy.RowMapping | sqlalchemy.engine.row.Row

• exclude_unset – bool Whether to exclude unset values.

Returns:

type: dict[str, Any], ModelProtocol]

Return type:

Union[