repositories

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.

order_by: Union[list[tuple[Union[str, InstrumentedAttribute[Any]], bool]], tuple[Union[str, InstrumentedAttribute[Any]], bool], None] = None

List or single OrderingPair to use for sorting.

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

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

model_type: type[TypeVar(ModelT, bound= base.ModelProtocol)]

The SQLAlchemy model class this repository manages.

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.

__init__(*args, **kwargs)

model_type: type[TypeVar(ModelT, bound= base.ModelProtocol)]

The SQLAlchemy model class this repository manages.

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 (Optional[ErrorMessages]) – A set of error messages to use for operations.

• **kwargs (Any) – Additional arguments.

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

error_messages: Optional[ErrorMessages] = None

async execute(statement)

Return type:

Result[Any]

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:

Row[Any]

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:

Optional[Row[Any]]

Returns:

The retrieved instance or None

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[Row[Any]]

Returns:

The list of instances, after filtering applied.

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

List records with total count.

Parameters:

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

• count_with_window_function (Optional[bool]) – Force list and count to use two queries instead of an analytical window function.

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

Return type:

tuple[list[Row[Any]], int]

Returns:

Count of records returned by query, ignoring pagination.

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

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

Repository for SQLAlchemy models.

Parameters:

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

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

• 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

• error_messages (Union[ErrorMessages, type[Empty], None]) – 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.

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

• **kwargs (Any) – Additional arguments.

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 (TypeVar(ModelT, bound= base.ModelProtocol)) – Instance to be added to the collection.

• 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

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

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[TypeVar(ModelT, bound= base.ModelProtocol)]) – list of Instances to be added to the collection.

• 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

Return type:

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

Returns:

The added instances.

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.

static check_not_found(item_or_none)

Raise advanced_alchemy.exceptions.NotFoundError if item_or_none is None.

Parameters:

item_or_none (Optional[TypeVar(ModelT, bound= base.ModelProtocol)]) – Item (T) to be tested for existence.

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

The item, if it exists.

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

Get the count of records returned by a query.

Parameters:

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

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

Return type:

int

Returns:

Count of records returned by query, ignoring pagination.

count_with_window_function: bool = True

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

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)

Delete instance identified by item_id.

Parameters:

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

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

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

Delete instance identified by item_id.

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:

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

Delete instances specified by referenced kwargs and filters.

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

Return type:

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

Returns:

The deleted instances.

error_messages: Optional[ErrorMessages] = None

Default error messages for the repository.

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

Return true if the object specified by kwargs exists.

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

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

Get instance identified by item_id.

Parameters:

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

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

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)

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

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 (Optional[bool]) – 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 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:

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.

classmethod get_id_attribute_value(item, id_attribute=None)

Get value of attribute named as id_attribute on item.

Parameters:

• item (Union[TypeVar(ModelT, bound= base.ModelProtocol), type[TypeVar(ModelT, bound= base.ModelProtocol)]]) – Anything that should have an attribute named as id_attribute value.

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

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

Get instance identified by kwargs.

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

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

Get instance identified by kwargs or None if not found.

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

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

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

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

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

• with_for_update (Optional[bool]) – 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 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:

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.

id_attribute: str = 'id'

Name of the unique identifier for the model.

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.

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

Get a list of instances, optionally filtered.

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[tuple[Union[str, InstrumentedAttribute[Any]], bool]], tuple[Union[str, InstrumentedAttribute[Any]], bool], 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:

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

Returns:

The list of instances, after filtering applied.

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

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[tuple[Union[str, InstrumentedAttribute[Any]], bool]], tuple[Union[str, InstrumentedAttribute[Any]], bool], 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[list[TypeVar(ModelT, bound= base.ModelProtocol)], int]

Returns:

Count of records returned by query, ignoring pagination.

loader_options: 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: Union[list[str], str, None] = None

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

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.

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 (Any) – Value of ID to be set on instance

• item (TypeVar(ModelT, bound= base.ModelProtocol)) – Anything that should have an attribute named as id_attribute value.

• id_attribute (Union[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:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

Item with item_id set to id_attribute

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

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

Update instance with the attribute values present on data.

Parameters:

• data (TypeVar(ModelT, bound= base.ModelProtocol)) – An instance that should have a value for self.id_attribute that exists in the collection.

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

• with_for_update (Optional[bool]) – 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.

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

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, uniquify=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[TypeVar(ModelT, bound= base.ModelProtocol)]) – A list of instances to update. Each should have a value for self.id_attribute that exists in the collection.

• 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

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

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

Returns:

The updated instances.

Raises:

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

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

Modify or create instance.

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

Parameters:

• data (TypeVar(ModelT, bound= base.ModelProtocol)) – 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 (Optional[Iterable[str]]) – an iterable of attribute names to pass into the update method.

• with_for_update (Optional[bool]) – 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.

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

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

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

Modify or create multiple instances.

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[TypeVar(ModelT, bound= base.ModelProtocol)]) – 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 (Optional[bool]) – Remove object from session before returning.

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

• no_merge (bool) – Skip the usage of optimized Merge statements

• match_fields (Union[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 (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:

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

Returns:

The updated or created instance.

Raises:

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

wrap_exceptions: bool = True

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

class advanced_alchemy.repository.SQLAlchemyAsyncRepositoryProtocol

Bases: FilterableRepositoryProtocol[ModelT], Protocol[ModelT]

Base Protocol

__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)

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

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

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

Return type:

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

async classmethod check_health(session)

Return type:

bool

static check_not_found(item_or_none)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

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

Return type:

int

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)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

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)

Return type:

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

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[TypeVar(ModelT, bound= base.ModelProtocol)]

error_messages: Optional[ErrorMessages] = None

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)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

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[TypeVar(ModelT, bound= base.ModelProtocol), bool]

classmethod get_id_attribute_value(item, id_attribute=None)

Return type:

Any

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:

TypeVar(ModelT, bound= base.ModelProtocol)

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:

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

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[TypeVar(ModelT, bound= base.ModelProtocol), bool]

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[TypeVar(ModelT, bound= base.ModelProtocol)]

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

Return type:

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

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

order_by: Union[list[tuple[Union[str, InstrumentedAttribute[Any]], bool]], tuple[Union[str, InstrumentedAttribute[Any]], bool], None] = None

classmethod set_id_attribute_value(item_id, item, id_attribute=None)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

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)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

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[TypeVar(ModelT, bound= base.ModelProtocol)]

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:

TypeVar(ModelT, bound= base.ModelProtocol)

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[TypeVar(ModelT, bound= base.ModelProtocol)]

wrap_exceptions: bool = True

id_attribute: str

statement: Select

session: Union[AsyncSession, async_scoped_session[AsyncSession]]

auto_expunge: bool

auto_refresh: bool

auto_commit: bool

class advanced_alchemy.repository.SQLAlchemyAsyncSlugRepository

Bases: SQLAlchemyAsyncRepository[ModelT], SQLAlchemyAsyncSlugRepositoryProtocol[ModelT]

Extends the repository to include slug model features..

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

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

Select record by slug value.

Return type:

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

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

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 (Union[ErrorMessages, type[Empty], None]) – Optional custom error message templates.

• 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]) – Specification for eager loading of relationships.

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

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

Returns:

The found model instance or None if not found.

Return type:

ModelT | None

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 (Optional[ErrorMessages]) – A set of error messages to use for operations.

• **kwargs (Any) – Additional arguments.

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.

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.

error_messages: Optional[ErrorMessages] = None

execute(statement)

Return type:

Result[Any]

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:

Row[Any]

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:

Optional[Row[Any]]

Returns:

The retrieved instance or None

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[Row[Any]]

Returns:

The list of instances, after filtering applied.

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

List records with total count.

Parameters:

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

• count_with_window_function (Optional[bool]) – Force list and count to use two queries instead of an analytical window function.

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

Return type:

tuple[list[Row[Any]], int]

Returns:

Count of records returned by query, ignoring pagination.

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

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

Repository for SQLAlchemy models.

Parameters:

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

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

• 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

• error_messages (Union[ErrorMessages, type[Empty], None]) – 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.

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

• **kwargs (Any) – Additional arguments.

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 (TypeVar(ModelT, bound= base.ModelProtocol)) – Instance to be added to the collection.

• 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

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

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[TypeVar(ModelT, bound= base.ModelProtocol)]) – list of Instances to be added to the collection.

• 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

Return type:

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

Returns:

The added instances.

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.

static check_not_found(item_or_none)

Raise advanced_alchemy.exceptions.NotFoundError if item_or_none is None.

Parameters:

item_or_none (Optional[TypeVar(ModelT, bound= base.ModelProtocol)]) – Item (T) to be tested for existence.

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

The item, if it exists.

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

Get the count of records returned by a query.

Parameters:

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

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

Return type:

int

Returns:

Count of records returned by query, ignoring pagination.

count_with_window_function: bool = True

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

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)

Delete instance identified by item_id.

Parameters:

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

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

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

Delete instance identified by item_id.

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:

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

Delete instances specified by referenced kwargs and filters.

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

Return type:

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

Returns:

The deleted instances.

error_messages: Optional[ErrorMessages] = None

Default error messages for the repository.

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

Return true if the object specified by kwargs exists.

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

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

Get instance identified by item_id.

Parameters:

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

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

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)

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

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 (Optional[bool]) – 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 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:

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.

classmethod get_id_attribute_value(item, id_attribute=None)

Get value of attribute named as id_attribute on item.

Parameters:

• item (Union[TypeVar(ModelT, bound= base.ModelProtocol), type[TypeVar(ModelT, bound= base.ModelProtocol)]]) – Anything that should have an attribute named as id_attribute value.

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

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

Get instance identified by kwargs.

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

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

Get instance identified by kwargs or None if not found.

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

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

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

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

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

• with_for_update (Optional[bool]) – 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 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:

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.

id_attribute: str = 'id'

Name of the unique identifier for the model.

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.

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

Get a list of instances, optionally filtered.

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[tuple[Union[str, InstrumentedAttribute[Any]], bool]], tuple[Union[str, InstrumentedAttribute[Any]], bool], 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:

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

Returns:

The list of instances, after filtering applied.

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

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[tuple[Union[str, InstrumentedAttribute[Any]], bool]], tuple[Union[str, InstrumentedAttribute[Any]], bool], 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[list[TypeVar(ModelT, bound= base.ModelProtocol)], int]

Returns:

Count of records returned by query, ignoring pagination.

loader_options: 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: Union[list[str], str, None] = None

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

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.

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 (Any) – Value of ID to be set on instance

• item (TypeVar(ModelT, bound= base.ModelProtocol)) – Anything that should have an attribute named as id_attribute value.

• id_attribute (Union[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:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

Item with item_id set to id_attribute

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

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

Update instance with the attribute values present on data.

Parameters:

• data (TypeVar(ModelT, bound= base.ModelProtocol)) – An instance that should have a value for self.id_attribute that exists in the collection.

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

• with_for_update (Optional[bool]) – 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.

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

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, uniquify=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[TypeVar(ModelT, bound= base.ModelProtocol)]) – A list of instances to update. Each should have a value for self.id_attribute that exists in the collection.

• 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

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

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

Returns:

The updated instances.

Raises:

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

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

Modify or create instance.

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

Parameters:

• data (TypeVar(ModelT, bound= base.ModelProtocol)) – 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 (Optional[Iterable[str]]) – an iterable of attribute names to pass into the update method.

• with_for_update (Optional[bool]) – 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.

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

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

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

Modify or create multiple instances.

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[TypeVar(ModelT, bound= base.ModelProtocol)]) – 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 (Optional[bool]) – Remove object from session before returning.

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

• no_merge (bool) – Skip the usage of optimized Merge statements

• match_fields (Union[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 (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:

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

Returns:

The updated or created instance.

Raises:

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

wrap_exceptions: bool = True

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

class advanced_alchemy.repository.SQLAlchemySyncRepositoryProtocol

Bases: FilterableRepositoryProtocol[ModelT], Protocol[ModelT]

Base Protocol

__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)

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

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

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

Return type:

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

classmethod check_health(session)

Return type:

bool

static check_not_found(item_or_none)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

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

Return type:

int

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)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

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)

Return type:

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

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[TypeVar(ModelT, bound= base.ModelProtocol)]

error_messages: Optional[ErrorMessages] = None

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)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

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[TypeVar(ModelT, bound= base.ModelProtocol), bool]

classmethod get_id_attribute_value(item, id_attribute=None)

Return type:

Any

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

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

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:

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

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[TypeVar(ModelT, bound= base.ModelProtocol), bool]

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[TypeVar(ModelT, bound= base.ModelProtocol)]

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

Return type:

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

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

order_by: Union[list[tuple[Union[str, InstrumentedAttribute[Any]], bool]], tuple[Union[str, InstrumentedAttribute[Any]], bool], None] = None

classmethod set_id_attribute_value(item_id, item, id_attribute=None)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

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)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

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[TypeVar(ModelT, bound= base.ModelProtocol)]

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:

TypeVar(ModelT, bound= base.ModelProtocol)

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[TypeVar(ModelT, bound= base.ModelProtocol)]

wrap_exceptions: bool = True

id_attribute: str

statement: Select

session: Union[Session, scoped_session[Session]]

auto_expunge: bool

auto_refresh: bool

auto_commit: bool

class advanced_alchemy.repository.SQLAlchemySyncSlugRepository

Bases: SQLAlchemySyncRepository[ModelT], SQLAlchemySyncSlugRepositoryProtocol[ModelT]

Extends the repository to include slug model features..

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

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

Select record by slug value.

Return type:

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

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

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 (Union[ErrorMessages, type[Empty], None]) – Optional custom error message templates.

• 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]) – Specification for eager loading of relationships.

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

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

Returns:

The found model instance or None if not found.

Return type:

ModelT | None

advanced_alchemy.repository.get_instrumented_attr(model, key)

Get an instrumented attribute from a model.

Parameters:

• model (type[ModelProtocol]) – SQLAlchemy model class.

• key (Union[str, InstrumentedAttribute[Any]]) – 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[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