repositories

final class advanced_alchemy.repository.Empty

Bases: object

A sentinel class used as placeholder.

class advanced_alchemy.repository.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.repository.FilterableRepository

Bases: FilterableRepositoryProtocol[ModelT]

Default implementation of a filterable repository.

Provides core filtering, ordering and pagination functionality for SQLAlchemy models.

Type Parameters:

ModelT: ModelProtocol The SQLAlchemy model type this repository handles.

model_type: type[ModelT]

The SQLAlchemy model class this repository manages.

prefer_any_dialects: tuple[str] | None = ('postgresql',)

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

order_by: list[OrderingPair] | OrderingPair | None = None

List or single OrderingPair to use for sorting.

class advanced_alchemy.repository.FilterableRepositoryProtocol

Bases: Protocol[ModelT]

Protocol defining the interface for filterable repositories.

This protocol defines the required attributes and methods that any filterable repository implementation must provide.

Type Parameters:

ModelT: ModelProtocol The SQLAlchemy model type this repository handles.

model_type

ModelProtocol The SQLAlchemy model class this repository manages.

model_type: type[ModelT]

__init__(*args, **kwargs)

class advanced_alchemy.repository.SQLAlchemyAsyncQueryRepository

Bases: object

SQLAlchemy Query Repository.

This is a loosely typed helper to query for when you need to select data in ways that don’t align to the normal repository pattern.

__init__(*, session, error_messages=None, **kwargs)

Repository pattern for SQLAlchemy models.

Parameters:

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

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

• **kwargs (Any) – Additional arguments.

error_messages: ErrorMessages | None = None

async get_one(statement, **kwargs)

Get instance identified by kwargs.

Parameters:

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

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

Return type:

RowMapping

Returns:

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

async get_one_or_none(statement, **kwargs)

Get instance identified by kwargs or None if not found.

Parameters:

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

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

Return type:

RowMapping | None

Returns:

The retrieved instance or None

async count(statement, **kwargs)

Get the count of records returned by a query.

Parameters:

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

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

Return type:

int

Returns:

Count of records returned by query, ignoring pagination.

async list_and_count(statement, force_basic_query_mode=None, **kwargs)

List records with total count.

Parameters:

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

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

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

Return type:

tuple[list[RowMapping], int]

Returns:

Count of records returned by query, ignoring pagination.

async list(statement, **kwargs)

Get a list of instances, optionally filtered.

Parameters:

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

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

Return type:

list[RowMapping]

Returns:

The list of instances, after filtering applied.

static check_not_found(item_or_none)

Raise NotFoundError if item_or_none is None.

Parameters:

item_or_none (Optional[TypeVar(T)]) – Item to be tested for existence.

Return type:

TypeVar(T)

Returns:

The item, if it exists.

async execute(statement)

Return type:

Result[Any]

class advanced_alchemy.repository.SQLAlchemyAsyncRepository

Bases: SQLAlchemyAsyncRepositoryProtocol[ModelT], FilterableRepository[ModelT]

Async SQLAlchemy repository implementation.

Provides a complete implementation of async database operations using SQLAlchemy, including CRUD operations, filtering, and relationship loading.

Type Parameters:

ModelT: The SQLAlchemy model type this repository handles.

See also

FilterableRepository

id_attribute: str = 'id'

Name of the unique identifier for the model.

loader_options: LoadSpec | None = None

Default loader options for the repository.

inherit_lazy_relationships: bool = True

Optionally ignore the default lazy configuration for model relationships. This is useful for when you want to replace instead of merge the model’s loaded relationships with the ones specified in the load or default_loader_options configuration.

merge_loader_options: bool = True

Merges the default loader options with the loader options specified in the load argument. This is useful for when you want to totally replace instead of merge the model’s loaded relationships with the ones specified in the load or default_loader_options configuration.

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 (...).

uniquify: bool = False

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

This is useful for certain SQLAlchemy uses cases such as applying contains_eager to a query containing a one-to-many relationship

__init__(*, statement=None, session, 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, wrap_exceptions=True, **kwargs)

Repository for SQLAlchemy models.

Parameters:

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

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

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

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

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

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

• wrap_exceptions (bool) – Wrap SQLAlchemy exceptions in a RepositoryError. When set to False, the original exception will be raised.

• **kwargs (Any) – Additional arguments.

error_messages: ErrorMessages | None = None

Default error messages for the repository.

wrap_exceptions: bool = True

Wrap SQLAlchemy exceptions in a RepositoryError. When set to False, the original exception will be raised.

classmethod get_id_attribute_value(item, id_attribute=None)

Get value of attribute named as id_attribute on item.

Parameters:

• item (ModelT | type[ModelT]) – Anything that should have an attribute named as id_attribute value.

• id_attribute (str | InstrumentedAttribute[Any] | None) – Allows customization of the unique identifier to use for model fetching. Defaults to None, but can reference any surrogate or candidate key for the table.

Return type:

Any

Returns:

The value of attribute on item named as id_attribute.

classmethod set_id_attribute_value(item_id, item, id_attribute=None)

Return the item after the ID is set to the appropriate attribute.

Parameters:

• item_id – Value of ID to be set on instance

• item – Anything that should have an attribute named as id_attribute value.

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

Returns:

Item with item_id set to id_attribute

static check_not_found(item_or_none)

Raise advanced_alchemy.exceptions.NotFoundError if item_or_none is None.

Parameters:

item_or_none (ModelT | None) – Item (T) to be tested for existence.

Return type:

ModelT

Returns:

The item, if it exists.

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

Add data to the collection.

Parameters:

• data (ModelT) – Instance to be added to the collection.

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

The added instance.

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

Add many data to the collection.

Parameters:

• data (list[ModelT]) – list of Instances to be added to the collection.

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

The added 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)

Delete instance identified by item_id.

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:

The deleted instance.

Raises:

NotFoundError – If no instance found identified by item_id.

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)

Delete instance identified by item_id.

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:

The deleted 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)

Delete instances specified by referenced kwargs and filters.

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) – Arguments to apply to a delete

Return type:

Sequence[ModelT]

Returns:

The deleted instances.

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

Return true if the object specified by kwargs exists.

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

bool

Returns:

True if the instance was found. False if not found..

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

Get instance identified by item_id.

Parameters:

• item_id – Identifier of the 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:

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

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

Get instance identified by kwargs.

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

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

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

Get instance identified by kwargs or None if not found.

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

The retrieved instance or None

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)

Get instance identified by kwargs or create if it doesn’t exist.

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, automatically perform an update operation on the model.

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

a tuple that includes the instance and whether it needed to be created. When using match_fields and actual model values differ from kwargs, the model value will be updated.

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)

Get instance identified by kwargs and update the model if the arguments are different.

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

a tuple that includes the instance and whether it needed to be updated. When using match_fields and actual model values differ from kwargs, the model value will be updated.

Raises:

NotFoundError – If no instance found identified by item_id.

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

Get the count of records returned by a query.

Parameters:

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

• 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) – Instance attribute value filters.

Return type:

int

Returns:

Count of records returned by query, ignoring pagination.

async update(data, *, 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)

Update instance with the attribute values present on data.

Parameters:

• data – An instance that should have a value for self.id_attribute that exists in the collection.

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

• execution_options – Set default execution options

Returns:

The updated instance.

Raises:

NotFoundError – If no instance found with same identifier as data.

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

Update one or more instances with the attribute values present on data.

This function has an optimized bulk update based on the configured SQL dialect: - For backends supporting RETURNING with executemany, a single bulk update with returning clause is executed. - For other backends, it does a bulk update and then returns the updated data after a refresh.

Parameters:

• data (list[ModelT]) – A list of instances to update. Each should have a value for self.id_attribute that exists in the collection.

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

list[ModelT]

Returns:

The updated instances.

Raises:

NotFoundError – If no instance found with same identifier as data.

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 records with total count.

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[list[ModelT], int]

Returns:

Count of records returned by query, ignoring pagination.

async upsert(data, *, 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)

Update or create instance.

Updates instance with the attribute values present on data, or creates a new instance if one doesn’t exist.

Parameters:

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

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

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

Return type:

ModelT

Returns:

The updated or created instance.

Raises:

NotFoundError – If no instance found with same identifier as data.

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)

Update or create instance.

Update instances with the attribute values present on data, or create a new instance if one doesn’t exist.

!!! tip

In most cases, you will want to set match_fields to the combination of attributes, excluded the primary key, that define uniqueness for a row.

Parameters:

• data (list[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 id_attribute.

• 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

• match_fields (list[str] | str | None) – a list of keys to use to match the existing model. When empty, automatically uses self.id_attribute (id by default) to match .

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

list[ModelT]

Returns:

The updated or created instance.

Raises:

NotFoundError – If no instance found with same identifier as data.

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

Get a list of instances, optionally filtered.

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

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

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

Return type:

list[ModelT]

Returns:

The list of instances, after filtering applied.

async classmethod check_health(session)

Perform a health check on the database.

Parameters:

session (Union[AsyncSession, async_scoped_session[AsyncSession]]) – through which we run a check statement

Return type:

bool

Returns:

True if healthy.

class advanced_alchemy.repository.SQLAlchemyAsyncRepositoryProtocol

Bases: FilterableRepositoryProtocol[ModelT], Protocol[ModelT]

Base Protocol

id_attribute: str

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

statement: Select[tuple[ModelT]]

session: AsyncSession | async_scoped_session[AsyncSession]

auto_expunge: bool

auto_refresh: bool

auto_commit: bool

order_by: list[OrderingPair] | OrderingPair | None = None

error_messages: ErrorMessages | None = None

wrap_exceptions: bool = True

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

classmethod get_id_attribute_value(item, id_attribute=None)

Return type:

Any

classmethod set_id_attribute_value(item_id, item, id_attribute=None)

static check_not_found(item_or_none)

Return type:

ModelT

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

Return type:

ModelT

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

Return type:

Sequence[ModelT]

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)

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)

async delete_where(*filters, auto_commit=None, auto_expunge=None, load=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, execution_options=None, sanity_check=True, **kwargs)

Return type:

Sequence[ModelT]

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

Return type:

bool

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

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

Return type:

ModelT

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

Return type:

ModelT | None

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)

Return type:

tuple[ModelT, bool]

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)

Return type:

tuple[ModelT, bool]

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

Return type:

int

async update(data, *, 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)

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

Return type:

list[ModelT]

async upsert(data, *, 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)

Return type:

ModelT

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)

Return type:

list[ModelT]

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

Return type:

tuple[list[ModelT], int]

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

Return type:

list[ModelT]

async classmethod check_health(session)

Return type:

bool

class advanced_alchemy.repository.SQLAlchemyAsyncSlugRepository

Bases: SQLAlchemyAsyncRepository[ModelT], SQLAlchemyAsyncSlugRepositoryProtocol[ModelT]

Extends the repository to include slug model features..

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

Select record by slug value.

Return type:

ModelT | None

async get_available_slug(value_to_slugify, **kwargs)

Get a unique slug for the supplied value.

If the value is found to exist, a random 4 digit character is appended to the end.

Override this method to change the default behavior

Parameters:

• value_to_slugify (str) – A string that should be converted to a unique slug.

• **kwargs (Any) – stuff

Returns:

a unique slug for the supplied value. This is safe for URLs and other unique identifiers.

Return type:

str

class advanced_alchemy.repository.SQLAlchemyAsyncSlugRepositoryProtocol

Bases: SQLAlchemyAsyncRepositoryProtocol[ModelT], Protocol[ModelT]

Protocol for SQLAlchemy repositories that support slug-based operations.

Extends the base repository protocol to add slug-related functionality.

Type Parameters:

ModelT: The SQLAlchemy model type this repository handles.

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

Get a model instance by its slug.

Parameters:

• slug (str) – The slug value to search for.

• error_messages (ErrorMessages | None | EmptyType) – Optional custom error message templates.

• load (LoadSpec | None) – Specification for eager loading of relationships.

• execution_options (dict[str, Any] | None) – Options for statement execution.

• **kwargs (Any) – Additional filtering criteria.

Returns:

The found model instance or None if not found.

Return type:

ModelT | None

async get_available_slug(value_to_slugify, **kwargs)

Generate a unique slug for a given value.

Parameters:

• value_to_slugify (str) – The string to convert to a slug.

• **kwargs (Any) – Additional parameters for slug generation.

Returns:

A unique slug derived from the input value.

Return type:

str

class advanced_alchemy.repository.SQLAlchemySyncQueryRepository

Bases: object

SQLAlchemy Query Repository.

This is a loosely typed helper to query for when you need to select data in ways that don’t align to the normal repository pattern.

__init__(*, session, error_messages=None, **kwargs)

Repository pattern for SQLAlchemy models.

Parameters:

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

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

• **kwargs (Any) – Additional arguments.

error_messages: ErrorMessages | None = None

get_one(statement, **kwargs)

Get instance identified by kwargs.

Parameters:

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

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

Return type:

RowMapping

Returns:

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

get_one_or_none(statement, **kwargs)

Get instance identified by kwargs or None if not found.

Parameters:

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

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

Return type:

RowMapping | None

Returns:

The retrieved instance or None

count(statement, **kwargs)

Get the count of records returned by a query.

Parameters:

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

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

Return type:

int

Returns:

Count of records returned by query, ignoring pagination.

list_and_count(statement, force_basic_query_mode=None, **kwargs)

List records with total count.

Parameters:

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

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

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

Return type:

tuple[list[RowMapping], int]

Returns:

Count of records returned by query, ignoring pagination.

list(statement, **kwargs)

Get a list of instances, optionally filtered.

Parameters:

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

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

Return type:

list[RowMapping]

Returns:

The list of instances, after filtering applied.

static check_not_found(item_or_none)

Raise NotFoundError if item_or_none is None.

Parameters:

item_or_none (Optional[TypeVar(T)]) – Item to be tested for existence.

Return type:

TypeVar(T)

Returns:

The item, if it exists.

execute(statement)

Return type:

Result[Any]

class advanced_alchemy.repository.SQLAlchemySyncRepository

Bases: SQLAlchemySyncRepositoryProtocol[ModelT], FilterableRepository[ModelT]

Async SQLAlchemy repository implementation.

Provides a complete implementation of async database operations using SQLAlchemy, including CRUD operations, filtering, and relationship loading.

Type Parameters:

ModelT: The SQLAlchemy model type this repository handles.

See also

FilterableRepository

id_attribute: str = 'id'

Name of the unique identifier for the model.

loader_options: LoadSpec | None = None

Default loader options for the repository.

inherit_lazy_relationships: bool = True

Optionally ignore the default lazy configuration for model relationships. This is useful for when you want to replace instead of merge the model’s loaded relationships with the ones specified in the load or default_loader_options configuration.

merge_loader_options: bool = True

Merges the default loader options with the loader options specified in the load argument. This is useful for when you want to totally replace instead of merge the model’s loaded relationships with the ones specified in the load or default_loader_options configuration.

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 (...).

uniquify: bool = False

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

This is useful for certain SQLAlchemy uses cases such as applying contains_eager to a query containing a one-to-many relationship

__init__(*, statement=None, session, 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, wrap_exceptions=True, **kwargs)

Repository for SQLAlchemy models.

Parameters:

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

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

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

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

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

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

• wrap_exceptions (bool) – Wrap SQLAlchemy exceptions in a RepositoryError. When set to False, the original exception will be raised.

• **kwargs (Any) – Additional arguments.

error_messages: ErrorMessages | None = None

Default error messages for the repository.

wrap_exceptions: bool = True

Wrap SQLAlchemy exceptions in a RepositoryError. When set to False, the original exception will be raised.

classmethod get_id_attribute_value(item, id_attribute=None)

Get value of attribute named as id_attribute on item.

Parameters:

• item (ModelT | type[ModelT]) – Anything that should have an attribute named as id_attribute value.

• id_attribute (str | InstrumentedAttribute[Any] | None) – Allows customization of the unique identifier to use for model fetching. Defaults to None, but can reference any surrogate or candidate key for the table.

Return type:

Any

Returns:

The value of attribute on item named as id_attribute.

classmethod set_id_attribute_value(item_id, item, id_attribute=None)

Return the item after the ID is set to the appropriate attribute.

Parameters:

• item_id – Value of ID to be set on instance

• item – Anything that should have an attribute named as id_attribute value.

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

Returns:

Item with item_id set to id_attribute

static check_not_found(item_or_none)

Raise advanced_alchemy.exceptions.NotFoundError if item_or_none is None.

Parameters:

item_or_none (ModelT | None) – Item (T) to be tested for existence.

Return type:

ModelT

Returns:

The item, if it exists.

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

Add data to the collection.

Parameters:

• data (ModelT) – Instance to be added to the collection.

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

The added instance.

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

Add many data to the collection.

Parameters:

• data (list[ModelT]) – list of Instances to be added to the collection.

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

The added 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)

Delete instance identified by item_id.

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:

The deleted instance.

Raises:

NotFoundError – If no instance found identified by item_id.

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)

Delete instance identified by item_id.

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:

The deleted 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)

Delete instances specified by referenced kwargs and filters.

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) – Arguments to apply to a delete

Return type:

Sequence[ModelT]

Returns:

The deleted instances.

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

Return true if the object specified by kwargs exists.

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

bool

Returns:

True if the instance was found. False if not found..

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

Get instance identified by item_id.

Parameters:

• item_id – Identifier of the 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:

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

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

Get instance identified by kwargs.

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

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

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

Get instance identified by kwargs or None if not found.

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

The retrieved instance or None

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)

Get instance identified by kwargs or create if it doesn’t exist.

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, automatically perform an update operation on the model.

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

a tuple that includes the instance and whether it needed to be created. When using match_fields and actual model values differ from kwargs, the model value will be updated.

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)

Get instance identified by kwargs and update the model if the arguments are different.

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

a tuple that includes the instance and whether it needed to be updated. When using match_fields and actual model values differ from kwargs, the model value will be updated.

Raises:

NotFoundError – If no instance found identified by item_id.

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

Get the count of records returned by a query.

Parameters:

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

• 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) – Instance attribute value filters.

Return type:

int

Returns:

Count of records returned by query, ignoring pagination.

update(data, *, 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)

Update instance with the attribute values present on data.

Parameters:

• data – An instance that should have a value for self.id_attribute that exists in the collection.

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

• execution_options – Set default execution options

Returns:

The updated instance.

Raises:

NotFoundError – If no instance found with same identifier as data.

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

Update one or more instances with the attribute values present on data.

This function has an optimized bulk update based on the configured SQL dialect: - For backends supporting RETURNING with executemany, a single bulk update with returning clause is executed. - For other backends, it does a bulk update and then returns the updated data after a refresh.

Parameters:

• data (list[ModelT]) – A list of instances to update. Each should have a value for self.id_attribute that exists in the collection.

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

list[ModelT]

Returns:

The updated instances.

Raises:

NotFoundError – If no instance found with same identifier as data.

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 records with total count.

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[list[ModelT], int]

Returns:

Count of records returned by query, ignoring pagination.

upsert(data, *, 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)

Update or create instance.

Updates instance with the attribute values present on data, or creates a new instance if one doesn’t exist.

Parameters:

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

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

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

Return type:

ModelT

Returns:

The updated or created instance.

Raises:

NotFoundError – If no instance found with same identifier as data.

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)

Update or create instance.

Update instances with the attribute values present on data, or create a new instance if one doesn’t exist.

!!! tip

In most cases, you will want to set match_fields to the combination of attributes, excluded the primary key, that define uniqueness for a row.

Parameters:

• data (list[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 id_attribute.

• 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

• match_fields (list[str] | str | None) – a list of keys to use to match the existing model. When empty, automatically uses self.id_attribute (id by default) to match .

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

list[ModelT]

Returns:

The updated or created instance.

Raises:

NotFoundError – If no instance found with same identifier as data.

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

Get a list of instances, optionally filtered.

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

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

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

Return type:

list[ModelT]

Returns:

The list of instances, after filtering applied.

classmethod check_health(session)

Perform a health check on the database.

Parameters:

session (Union[Session, scoped_session[Session]]) – through which we run a check statement

Return type:

bool

Returns:

True if healthy.

class advanced_alchemy.repository.SQLAlchemySyncRepositoryProtocol

Bases: FilterableRepositoryProtocol[ModelT], Protocol[ModelT]

Base Protocol

id_attribute: str

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

statement: Select[tuple[ModelT]]

session: Session | scoped_session[Session]

auto_expunge: bool

auto_refresh: bool

auto_commit: bool

order_by: list[OrderingPair] | OrderingPair | None = None

error_messages: ErrorMessages | None = None

wrap_exceptions: bool = True

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

classmethod get_id_attribute_value(item, id_attribute=None)

Return type:

Any

classmethod set_id_attribute_value(item_id, item, id_attribute=None)

static check_not_found(item_or_none)

Return type:

ModelT

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

Return type:

ModelT

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

Return type:

Sequence[ModelT]

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)

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)

delete_where(*filters, auto_commit=None, auto_expunge=None, load=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, execution_options=None, sanity_check=True, **kwargs)

Return type:

Sequence[ModelT]

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

Return type:

bool

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

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

Return type:

ModelT

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

Return type:

ModelT | None

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)

Return type:

tuple[ModelT, bool]

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)

Return type:

tuple[ModelT, bool]

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

Return type:

int

update(data, *, 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)

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

Return type:

list[ModelT]

upsert(data, *, 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)

Return type:

ModelT

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)

Return type:

list[ModelT]

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

Return type:

tuple[list[ModelT], int]

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

Return type:

list[ModelT]

classmethod check_health(session)

Return type:

bool

class advanced_alchemy.repository.SQLAlchemySyncSlugRepository

Bases: SQLAlchemySyncRepository[ModelT], SQLAlchemySyncSlugRepositoryProtocol[ModelT]

Extends the repository to include slug model features..

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

Select record by slug value.

Return type:

ModelT | None

get_available_slug(value_to_slugify, **kwargs)

Get a unique slug for the supplied value.

If the value is found to exist, a random 4 digit character is appended to the end.

Override this method to change the default behavior

Parameters:

• value_to_slugify (str) – A string that should be converted to a unique slug.

• **kwargs (Any) – stuff

Returns:

a unique slug for the supplied value. This is safe for URLs and other unique identifiers.

Return type:

str

class advanced_alchemy.repository.SQLAlchemySyncSlugRepositoryProtocol

Bases: SQLAlchemySyncRepositoryProtocol[ModelT], Protocol[ModelT]

Protocol for SQLAlchemy repositories that support slug-based operations.

Extends the base repository protocol to add slug-related functionality.

Type Parameters:

ModelT: The SQLAlchemy model type this repository handles.

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

Get a model instance by its slug.

Parameters:

• slug (str) – The slug value to search for.

• error_messages (ErrorMessages | None | EmptyType) – Optional custom error message templates.

• load (LoadSpec | None) – Specification for eager loading of relationships.

• execution_options (dict[str, Any] | None) – Options for statement execution.

• **kwargs (Any) – Additional filtering criteria.

Returns:

The found model instance or None if not found.

Return type:

ModelT | None

get_available_slug(value_to_slugify, **kwargs)

Generate a unique slug for a given value.

Parameters:

• value_to_slugify (str) – The string to convert to a slug.

• **kwargs (Any) – Additional parameters for slug generation.

Returns:

A unique slug derived from the input value.

Return type:

str

advanced_alchemy.repository.get_instrumented_attr(model, key)

Get an instrumented attribute from a model.

Parameters:

• model – SQLAlchemy model class.

• key – Either a string attribute name or an sqlalchemy.orm.InstrumentedAttribute.

Returns:

The instrumented attribute from the model.

Return type:

sqlalchemy.orm.InstrumentedAttribute

advanced_alchemy.repository.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