repositories

class advanced_alchemy.repository.SQLAlchemyAsyncRepository

Bases: SQLAlchemyAsyncRepositoryProtocol[ModelT], FilterableRepository[ModelT]

SQLAlchemy based implementation of the repository interface.

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

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

__init__(*, statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, session: AsyncSession | async_scoped_session[AsyncSession], auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: list[OrderingPair] | OrderingPair | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None, **kwargs: Any) → None

Repository pattern for SQLAlchemy models.

Parameters:

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

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

• auto_expunge – Remove object from session before returning.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

• order_by – Set default order options for queries.

• load – Set default relationships to be loaded

• execution_options – Set default execution options

• error_messages – A set of custom error messages to use for operations

• **kwargs – Additional arguments.

classmethod get_id_attribute_value(item: ModelT | type[ModelT], id_attribute: str | InstrumentedAttribute[Any] | None = None) → Any

Get value of attribute named as id_attribute on item.

Parameters:

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

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

Returns:

The value of attribute on item named as id_attribute.

classmethod set_id_attribute_value(item_id: Any, item: ModelT, id_attribute: str | InstrumentedAttribute[Any] | None = None) → ModelT

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

Parameters:

• item_id – Value of ID to be set on instance

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

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

Returns:

Item with item_id set to id_attribute

static check_not_found(item_or_none: ModelT | None) → ModelT

Raise advanced_alchemy.exceptions.NotFoundError if item_or_none is None.

Parameters:

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

Returns:

The item, if it exists.

async add(data: ~advanced_alchemy.repository.typing.ModelT, *, auto_commit: bool | None = None, auto_expunge: bool | None = None, auto_refresh: bool | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>) → ModelT

Add data to the collection.

Parameters:

• data – Instance to be added to the collection.

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

• auto_refresh – Refresh object from session before returning. Defaults to SQLAlchemyAsyncRepository.auto_refresh

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

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

Returns:

The added instance.

async add_many(data: list[~ModelT], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>) → Sequence[ModelT]

Add many data to the collection.

Parameters:

• data – list of Instances to be added to the collection.

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

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

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

Returns:

The added instances.

async delete(item_id: ~typing.Any, *, auto_commit: bool | None = None, auto_expunge: bool | None = None, id_attribute: str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, typing.Any] | None = None) → ModelT

Delete instance identified by item_id.

Parameters:

• item_id – Identifier of instance to be deleted.

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

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

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

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

• load – Set default relationships to be loaded

• execution_options – Set default execution options

Returns:

The deleted instance.

Raises:

NotFoundError – If no instance found identified by item_id.

async delete_many(item_ids: list[typing.Any], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, id_attribute: str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | None = None, chunk_size: int | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, typing.Any] | None = None) → Sequence[ModelT]

Delete instance identified by item_id.

Parameters:

• item_ids – Identifier of instance to be deleted.

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

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

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

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

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

• load – Set default relationships to be loaded

• execution_options – Set default execution options

Returns:

The deleted instances.

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

Delete instances specified by referenced kwargs and filters.

Parameters:

• *filters – Types for specific filtering operations.

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

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

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

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

• load – Set default relationships to be loaded

• execution_options – Set default execution options

• **kwargs – Arguments to apply to a delete

Returns:

The deleted instances.

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

Return true if the object specified by kwargs exists.

Parameters:

• *filters – Types for specific filtering operations.

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

• load – Set default relationships to be loaded

• execution_options – Set default execution options

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

Returns:

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

async get(item_id: ~typing.Any, *, auto_expunge: bool | None = None, statement: ~sqlalchemy.sql.selectable.Select | ~sqlalchemy.sql.lambdas.StatementLambdaElement | None = None, id_attribute: str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, typing.Any] | None = None) → ModelT

Get instance identified by item_id.

Parameters:

• item_id – Identifier of the instance to be retrieved.

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

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

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

Returns:

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

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

Get instance identified by kwargs.

Parameters:

• *filters – Types for specific filtering operations.

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

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

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

Returns:

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

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

Get instance identified by kwargs or None if not found.

Parameters:

• *filters – Types for specific filtering operations.

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

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

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

Returns:

The retrieved instance or None

async get_or_upsert(*filters: StatementFilter | ColumnElement[bool], match_fields: list[str] | str | None = None, upsert: bool = True, attribute_names: Iterable[str] | None = None, with_for_update: bool | None = None, auto_commit: bool | None = None, auto_expunge: bool | None = None, auto_refresh: bool | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None, **kwargs: Any) → tuple[ModelT, bool]

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

Parameters:

• *filters – Types for specific filtering operations.

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

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

• attribute_names – an iterable of attribute names to pass into the update method.

• with_for_update – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT

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

• auto_refresh – Refresh object from session before returning. Defaults to SQLAlchemyAsyncRepository.auto_refresh

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

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

Returns:

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

async get_and_update(*filters: StatementFilter | ColumnElement[bool], match_fields: list[str] | str | None = None, attribute_names: Iterable[str] | None = None, with_for_update: bool | None = None, auto_commit: bool | None = None, auto_expunge: bool | None = None, auto_refresh: bool | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None, **kwargs: Any) → tuple[ModelT, bool]

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

Parameters:

• *filters – Types for specific filtering operations.

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

• attribute_names – an iterable of attribute names to pass into the update method.

• with_for_update – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT

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

• auto_refresh – Refresh object from session before returning. Defaults to SQLAlchemyAsyncRepository.auto_refresh

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

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

Returns:

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

Raises:

NotFoundError – If no instance found identified by item_id.

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

Get the count of records returned by a query.

Parameters:

• *filters – Types for specific filtering operations.

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

• **kwargs – Instance attribute value filters.

Returns:

Count of records returned by query, ignoring pagination.

async update(data: ~advanced_alchemy.repository.typing.ModelT, *, attribute_names: ~typing.Iterable[str] | None = None, with_for_update: bool | None = None, auto_commit: bool | None = None, auto_expunge: bool | None = None, auto_refresh: bool | None = None, id_attribute: str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, typing.Any] | None = None) → ModelT

Update instance with the attribute values present on data.

Parameters:

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

• attribute_names – an iterable of attribute names to pass into the update method.

• with_for_update – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT

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

• auto_refresh – Refresh object from session before returning. Defaults to SQLAlchemyAsyncRepository.auto_refresh

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

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

Returns:

The updated instance.

Raises:

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

async update_many(data: list[~ModelT], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, typing.Any] | None = None) → list[ModelT]

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 – A list of instances to update. Each should have a value for self.id_attribute that exists in the collection.

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

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

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

• load – Set default relationships to be loaded

• execution_options – Set default execution options

Returns:

The updated instances.

Raises:

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

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

List records with total count.

Parameters:

• *filters – Types for specific filtering operations.

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

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

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

• order_by – Set default order options for queries.

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

• **kwargs – Instance attribute value filters.

Returns:

Count of records returned by query, ignoring pagination.

async upsert(data: ~advanced_alchemy.repository.typing.ModelT, *, attribute_names: ~typing.Iterable[str] | None = None, with_for_update: bool | None = None, auto_expunge: bool | None = None, auto_commit: bool | None = None, auto_refresh: bool | None = None, match_fields: list[str] | str | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, typing.Any] | None = None) → ModelT

Update or create instance.

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

Parameters:

• data – 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 – an iterable of attribute names to pass into the update method.

• with_for_update – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT

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

• auto_refresh – Refresh object from session before returning. Defaults to SQLAlchemyAsyncRepository.auto_refresh

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

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

Returns:

The updated or created instance.

Raises:

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

async upsert_many(data: list[~ModelT], *, auto_expunge: bool | None = None, auto_commit: bool | None = None, no_merge: bool = False, match_fields: list[str] | str | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, typing.Any] | None = None) → list[ModelT]

Update or create instance.

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

!!! tip

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

Parameters:

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

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

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

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

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

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

• load – Set default relationships to be loaded

• execution_options – Set default execution options

Returns:

The updated or created instance.

Raises:

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

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

Get a list of instances, optionally filtered.

Parameters:

• *filters – Types for specific filtering operations.

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

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

• order_by – Set default order options for queries.

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

• **kwargs – Instance attribute value filters.

Returns:

The list of instances, after filtering applied.

filter_collection_by_kwargs(collection: Select | StatementLambdaElement, /, **kwargs: Any) → StatementLambdaElement

Filter the collection by kwargs.

Parameters:

• collection – statement to filter

• **kwargs – key/value pairs such that objects remaining in the collection after filtering have the property that their attribute named key has value equal to value.

async classmethod check_health(session: AsyncSession | async_scoped_session[AsyncSession]) → bool

Perform a health check on the database.

Parameters:

session – through which we run a check statement

Returns:

True if healthy.

class advanced_alchemy.repository.SQLAlchemyAsyncRepositoryProtocol

Bases: FilterableRepositoryProtocol[ModelT], Protocol[ModelT]

Base Protocol

__init__(*, statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, session: AsyncSession | async_scoped_session[AsyncSession], auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None, order_by: list[OrderingPair] | OrderingPair | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, **kwargs: Any) → None

class advanced_alchemy.repository.SQLAlchemyAsyncSlugRepositoryProtocol

Bases: SQLAlchemyAsyncRepositoryProtocol[ModelT], Protocol[ModelT]

class advanced_alchemy.repository.FilterableRepositoryProtocol

Bases: Protocol[ModelT]

__init__(*args, **kwargs)

class advanced_alchemy.repository.SQLAlchemySyncRepositoryProtocol

Bases: FilterableRepositoryProtocol[ModelT], Protocol[ModelT]

Base Protocol

__init__(*, statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, session: Session | scoped_session[Session], auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None, order_by: list[OrderingPair] | OrderingPair | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, **kwargs: Any) → None

class advanced_alchemy.repository.SQLAlchemySyncSlugRepositoryProtocol

Bases: SQLAlchemySyncRepositoryProtocol[ModelT], Protocol[ModelT]

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: AsyncSession | async_scoped_session[AsyncSession], error_messages: ErrorMessages | None = None, **kwargs: Any) → None

Repository pattern for SQLAlchemy models.

Parameters:

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

• error_messages – A set of error messages to use for operations.

• **kwargs – Additional arguments.

async get_one(statement: Select, **kwargs: Any) → RowMapping

Get instance identified by kwargs.

Parameters:

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

• **kwargs – Instance attribute value filters.

Returns:

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

async get_one_or_none(statement: Select, **kwargs: Any) → RowMapping | None

Get instance identified by kwargs or None if not found.

Parameters:

statement – To facilitate customization of the underlying select query. **kwargs: Instance attribute value filters.

Returns:

The retrieved instance or None

async count(statement: Select, **kwargs: Any) → int

Get the count of records returned by a query.

Parameters:

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

• **kwargs – Instance attribute value filters.

Returns:

Count of records returned by query, ignoring pagination.

async list_and_count(statement: Select, force_basic_query_mode: bool | None = None, **kwargs: Any) → tuple[list[sqlalchemy.engine.row.RowMapping], int]

List records with total count.

Parameters:

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

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

• **kwargs – Instance attribute value filters.

Returns:

Count of records returned by query, ignoring pagination.

async list(statement: Select, **kwargs: Any) → list[sqlalchemy.engine.row.RowMapping]

Get a list of instances, optionally filtered.

Parameters:

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

• **kwargs – Instance attribute value filters.

Returns:

The list of instances, after filtering applied.

static check_not_found(item_or_none: T | None) → T

Raise RepositoryNotFoundException if item_or_none is None.

Parameters:

item_or_none – Item to be tested for existence.

Returns:

The item, if it exists.

class advanced_alchemy.repository.SQLAlchemyAsyncSlugRepository

Bases: SQLAlchemyAsyncRepository[ModelT], SQLAlchemyAsyncSlugRepositoryProtocol[ModelT]

Extends the repository to include slug model features..

async get_by_slug(slug: str, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, typing.Any] | None = None, **kwargs: ~typing.Any) → ModelT | None

Select record by slug value.

async get_available_slug(value_to_slugify: str, **kwargs: Any) → str

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 – stuff

Returns:

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

Return type:

str

class advanced_alchemy.repository.SQLAlchemySyncSlugRepository

Bases: SQLAlchemySyncRepository[ModelT], SQLAlchemySyncSlugRepositoryProtocol[ModelT]

Extends the repository to include slug model features..

get_by_slug(slug: str, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, typing.Any] | None = None, **kwargs: ~typing.Any) → ModelT | None

Select record by slug value.

get_available_slug(value_to_slugify: str, **kwargs: Any) → str

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 – stuff

Returns:

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

Return type:

str

class advanced_alchemy.repository.SQLAlchemySyncRepository

Bases: SQLAlchemySyncRepositoryProtocol[ModelT], FilterableRepository[ModelT]

SQLAlchemy based implementation of the repository interface.

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

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

__init__(*, statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, session: Session | scoped_session[Session], auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: list[OrderingPair] | OrderingPair | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None, **kwargs: Any) → None

Repository pattern for SQLAlchemy models.

Parameters:

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

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

• auto_expunge – Remove object from session before returning.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

• order_by – Set default order options for queries.

• load – Set default relationships to be loaded

• execution_options – Set default execution options

• error_messages – A set of custom error messages to use for operations

• **kwargs – Additional arguments.

classmethod get_id_attribute_value(item: ModelT | type[ModelT], id_attribute: str | InstrumentedAttribute[Any] | None = None) → Any

Get value of attribute named as id_attribute on item.

Parameters:

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

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

Returns:

The value of attribute on item named as id_attribute.

classmethod set_id_attribute_value(item_id: Any, item: ModelT, id_attribute: str | InstrumentedAttribute[Any] | None = None) → ModelT

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

Parameters:

• item_id – Value of ID to be set on instance

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

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

Returns:

Item with item_id set to id_attribute

static check_not_found(item_or_none: ModelT | None) → ModelT

Raise advanced_alchemy.exceptions.NotFoundError if item_or_none is None.

Parameters:

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

Returns:

The item, if it exists.

add(data: ~advanced_alchemy.repository.typing.ModelT, *, auto_commit: bool | None = None, auto_expunge: bool | None = None, auto_refresh: bool | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>) → ModelT

Add data to the collection.

Parameters:

• data – Instance to be added to the collection.

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

• auto_refresh – Refresh object from session before returning. Defaults to SQLAlchemyAsyncRepository.auto_refresh

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

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

Returns:

The added instance.

add_many(data: list[~ModelT], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>) → Sequence[ModelT]

Add many data to the collection.

Parameters:

• data – list of Instances to be added to the collection.

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

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

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

Returns:

The added instances.

delete(item_id: ~typing.Any, *, auto_commit: bool | None = None, auto_expunge: bool | None = None, id_attribute: str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, typing.Any] | None = None) → ModelT

Delete instance identified by item_id.

Parameters:

• item_id – Identifier of instance to be deleted.

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

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

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

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

• load – Set default relationships to be loaded

• execution_options – Set default execution options

Returns:

The deleted instance.

Raises:

NotFoundError – If no instance found identified by item_id.

delete_many(item_ids: list[typing.Any], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, id_attribute: str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | None = None, chunk_size: int | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, typing.Any] | None = None) → Sequence[ModelT]

Delete instance identified by item_id.

Parameters:

• item_ids – Identifier of instance to be deleted.

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

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

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

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

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

• load – Set default relationships to be loaded

• execution_options – Set default execution options

Returns:

The deleted instances.

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

Delete instances specified by referenced kwargs and filters.

Parameters:

• *filters – Types for specific filtering operations.

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

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

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

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

• load – Set default relationships to be loaded

• execution_options – Set default execution options

• **kwargs – Arguments to apply to a delete

Returns:

The deleted instances.

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

Return true if the object specified by kwargs exists.

Parameters:

• *filters – Types for specific filtering operations.

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

• load – Set default relationships to be loaded

• execution_options – Set default execution options

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

Returns:

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

get(item_id: ~typing.Any, *, auto_expunge: bool | None = None, statement: ~sqlalchemy.sql.selectable.Select | ~sqlalchemy.sql.lambdas.StatementLambdaElement | None = None, id_attribute: str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, typing.Any] | None = None) → ModelT

Get instance identified by item_id.

Parameters:

• item_id – Identifier of the instance to be retrieved.

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

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

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

Returns:

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

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

Get instance identified by kwargs.

Parameters:

• *filters – Types for specific filtering operations.

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

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

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

Returns:

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

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

Get instance identified by kwargs or None if not found.

Parameters:

• *filters – Types for specific filtering operations.

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

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

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

Returns:

The retrieved instance or None

get_or_upsert(*filters: StatementFilter | ColumnElement[bool], match_fields: list[str] | str | None = None, upsert: bool = True, attribute_names: Iterable[str] | None = None, with_for_update: bool | None = None, auto_commit: bool | None = None, auto_expunge: bool | None = None, auto_refresh: bool | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None, **kwargs: Any) → tuple[ModelT, bool]

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

Parameters:

• *filters – Types for specific filtering operations.

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

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

• attribute_names – an iterable of attribute names to pass into the update method.

• with_for_update – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT

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

• auto_refresh – Refresh object from session before returning. Defaults to SQLAlchemyAsyncRepository.auto_refresh

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

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

Returns:

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

get_and_update(*filters: StatementFilter | ColumnElement[bool], match_fields: list[str] | str | None = None, attribute_names: Iterable[str] | None = None, with_for_update: bool | None = None, auto_commit: bool | None = None, auto_expunge: bool | None = None, auto_refresh: bool | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None, **kwargs: Any) → tuple[ModelT, bool]

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

Parameters:

• *filters – Types for specific filtering operations.

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

• attribute_names – an iterable of attribute names to pass into the update method.

• with_for_update – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT

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

• auto_refresh – Refresh object from session before returning. Defaults to SQLAlchemyAsyncRepository.auto_refresh

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

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

Returns:

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

Raises:

NotFoundError – If no instance found identified by item_id.

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

Get the count of records returned by a query.

Parameters:

• *filters – Types for specific filtering operations.

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

• **kwargs – Instance attribute value filters.

Returns:

Count of records returned by query, ignoring pagination.

update(data: ~advanced_alchemy.repository.typing.ModelT, *, attribute_names: ~typing.Iterable[str] | None = None, with_for_update: bool | None = None, auto_commit: bool | None = None, auto_expunge: bool | None = None, auto_refresh: bool | None = None, id_attribute: str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, typing.Any] | None = None) → ModelT

Update instance with the attribute values present on data.

Parameters:

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

• attribute_names – an iterable of attribute names to pass into the update method.

• with_for_update – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT

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

• auto_refresh – Refresh object from session before returning. Defaults to SQLAlchemyAsyncRepository.auto_refresh

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

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

Returns:

The updated instance.

Raises:

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

update_many(data: list[~ModelT], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, typing.Any] | None = None) → list[ModelT]

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 – A list of instances to update. Each should have a value for self.id_attribute that exists in the collection.

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

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

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

• load – Set default relationships to be loaded

• execution_options – Set default execution options

Returns:

The updated instances.

Raises:

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

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

List records with total count.

Parameters:

• *filters – Types for specific filtering operations.

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

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

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

• order_by – Set default order options for queries.

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

• **kwargs – Instance attribute value filters.

Returns:

Count of records returned by query, ignoring pagination.

upsert(data: ~advanced_alchemy.repository.typing.ModelT, *, attribute_names: ~typing.Iterable[str] | None = None, with_for_update: bool | None = None, auto_expunge: bool | None = None, auto_commit: bool | None = None, auto_refresh: bool | None = None, match_fields: list[str] | str | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, typing.Any] | None = None) → ModelT

Update or create instance.

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

Parameters:

• data – 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 – an iterable of attribute names to pass into the update method.

• with_for_update – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT

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

• auto_refresh – Refresh object from session before returning. Defaults to SQLAlchemyAsyncRepository.auto_refresh

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

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

Returns:

The updated or created instance.

Raises:

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

upsert_many(data: list[~ModelT], *, auto_expunge: bool | None = None, auto_commit: bool | None = None, no_merge: bool = False, match_fields: list[str] | str | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, typing.Any] | None = None) → list[ModelT]

Update or create instance.

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

!!! tip

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

Parameters:

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

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

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

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

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

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

• load – Set default relationships to be loaded

• execution_options – Set default execution options

Returns:

The updated or created instance.

Raises:

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

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

Get a list of instances, optionally filtered.

Parameters:

• *filters – Types for specific filtering operations.

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

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

• order_by – Set default order options for queries.

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

• **kwargs – Instance attribute value filters.

Returns:

The list of instances, after filtering applied.

filter_collection_by_kwargs(collection: Select | StatementLambdaElement, /, **kwargs: Any) → StatementLambdaElement

Filter the collection by kwargs.

Parameters:

• collection – statement to filter

• **kwargs – key/value pairs such that objects remaining in the collection after filtering have the property that their attribute named key has value equal to value.

classmethod check_health(session: Session | scoped_session[Session]) → bool

Perform a health check on the database.

Parameters:

session – through which we run a check statement

Returns:

True if healthy.

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: Session | scoped_session[Session], error_messages: ErrorMessages | None = None, **kwargs: Any) → None

Repository pattern for SQLAlchemy models.

Parameters:

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

• error_messages – A set of error messages to use for operations.

• **kwargs – Additional arguments.

get_one(statement: Select, **kwargs: Any) → RowMapping

Get instance identified by kwargs.

Parameters:

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

• **kwargs – Instance attribute value filters.

Returns:

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

get_one_or_none(statement: Select, **kwargs: Any) → RowMapping | None

Get instance identified by kwargs or None if not found.

Parameters:

statement – To facilitate customization of the underlying select query. **kwargs: Instance attribute value filters.

Returns:

The retrieved instance or None

count(statement: Select, **kwargs: Any) → int

Get the count of records returned by a query.

Parameters:

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

• **kwargs – Instance attribute value filters.

Returns:

Count of records returned by query, ignoring pagination.

list_and_count(statement: Select, force_basic_query_mode: bool | None = None, **kwargs: Any) → tuple[list[sqlalchemy.engine.row.RowMapping], int]

List records with total count.

Parameters:

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

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

• **kwargs – Instance attribute value filters.

Returns:

Count of records returned by query, ignoring pagination.

list(statement: Select, **kwargs: Any) → list[sqlalchemy.engine.row.RowMapping]

Get a list of instances, optionally filtered.

Parameters:

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

• **kwargs – Instance attribute value filters.

Returns:

The list of instances, after filtering applied.

static check_not_found(item_or_none: T | None) → T

Raise RepositoryNotFoundException if item_or_none is None.

Parameters:

item_or_none – Item to be tested for existence.

Returns:

The item, if it exists.

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

Return ORM Object from Dictionary.

final class advanced_alchemy.repository.Empty

Bases: object

A sentinel class used as placeholder.

class advanced_alchemy.repository.ErrorMessages

Bases: TypedDict