repositories

final class advanced_alchemy.repository.Empty

Bases: object

A sentinel class used as placeholder.

class advanced_alchemy.repository.ErrorMessages

Bases: TypedDict

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

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

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

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

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

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

class advanced_alchemy.repository.FilterableRepository

Bases: FilterableRepositoryProtocol[ModelT]

Default implementation of a filterable repository.

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

Type Parameters:

ModelT: ModelProtocol The SQLAlchemy model type this repository handles.

model_type: type[ModelT]

The SQLAlchemy model class this repository manages.

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

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

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

List or single OrderingPair to use for sorting.

class advanced_alchemy.repository.FilterableRepositoryProtocol

Bases: Protocol[ModelT]

Protocol defining the interface for filterable repositories.

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

Type Parameters:

ModelT: ModelProtocol The SQLAlchemy model type this repository handles.

model_type

ModelProtocol The SQLAlchemy model class this repository manages.

Type:

type[ModelT]

model_type: type[ModelT]

__init__(*args, **kwargs)

class advanced_alchemy.repository.SQLAlchemyAsyncQueryRepository

Bases: object

SQLAlchemy Query Repository.

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

__init__(*, session: 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.

error_messages: ErrorMessages | None = None

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

Get instance identified by kwargs.

Parameters:

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

• **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[RowMapping], int]

List records with total count.

Parameters:

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

• 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[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 NotFoundError if item_or_none is None.

Parameters:

item_or_none – Item to be tested for existence.

Returns:

The item, if it exists.

async execute(statement: ReturningDelete[tuple[Any]] | ReturningUpdate[tuple[Any]] | Select[tuple[Any]] | Update | Delete | Select[Any]) → Result[Any]

class advanced_alchemy.repository.SQLAlchemyAsyncRepository

Bases: SQLAlchemyAsyncRepositoryProtocol[ModelT], FilterableRepository[ModelT]

Async SQLAlchemy repository implementation.

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

Type Parameters:

ModelT: The SQLAlchemy model type this repository handles.

See also

FilterableRepository

id_attribute: str = 'id'

Name of the unique identifier for the model.

loader_options: LoadSpec | None = None

Default loader options for the repository.

inherit_lazy_relationships: bool = True

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

merge_loader_options: bool = True

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

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

Default execution options for the repository.

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

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

uniquify: bool = False

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

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

__init__(*, statement: Select[tuple[ModelT]] | 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 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.

error_messages: ErrorMessages | None = None

Default error messages for the repository.

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'>) → advanced_alchemy.repository.typing.ModelT

Add data to the collection.

Parameters:

• data – Instance to be added to the collection.

• auto_expunge – Remove object from session before returning.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

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

Returns:

The added instance.

async add_many(data: list[advanced_alchemy.repository.typing.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[advanced_alchemy.repository.typing.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.

• auto_commit – Commit objects before returning.

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

Returns:

The added instances.

async delete(item_id: Any, *, auto_commit: bool | None = None, auto_expunge: bool | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, 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.

• auto_commit – Commit objects before returning.

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

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

• load – Set default relationships to be loaded

• execution_options – Set default execution options

Returns:

The deleted instance.

Raises:

NotFoundError – If no instance found identified by item_id.

async delete_many(item_ids: list[Any], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, chunk_size: int | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, 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.

• auto_commit – Commit objects before returning.

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

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

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

• load – Set default relationships to be loaded

• execution_options – Set default execution options

Returns:

The deleted instances.

async delete_where(*filters: 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.

• auto_commit – Commit objects before returning.

• 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: Any, *, auto_expunge: bool | None = None, statement: Select[tuple[ModelT]] | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, 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.

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

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

Returns:

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

async get_one(*filters: StatementFilter | ColumnElement[bool], auto_expunge: bool | None = None, statement: Select[tuple[ModelT]] | 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.

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

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

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

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

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

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

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

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

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

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

• 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: ModelT, *, 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, id_attribute: str | InstrumentedAttribute[Any] | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, 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.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

Returns:

The updated instance.

Raises:

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

async update_many(data: list[advanced_alchemy.repository.typing.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[advanced_alchemy.repository.typing.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.

• auto_commit – Commit objects before returning.

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

• load – Set default relationships to be loaded

• execution_options – Set default execution options

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

• auto_expunge – Remove object from session before returning.

• 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) → advanced_alchemy.repository.typing.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.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

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

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

• load – Set 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[advanced_alchemy.repository.typing.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[advanced_alchemy.repository.typing.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.

• auto_commit – Commit objects before returning.

• no_merge – Skip the usage of optimized Merge statements

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

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

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

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

id_attribute: str

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

statement: Select[tuple[ModelT]]

session: AsyncSession | async_scoped_session[AsyncSession]

auto_expunge: bool

auto_refresh: bool

auto_commit: bool

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

error_messages: ErrorMessages | None = None

__init__(*, statement: Select[tuple[ModelT]] | 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

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

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

static check_not_found(item_or_none: ModelT | None) → ModelT

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'>) → advanced_alchemy.repository.typing.ModelT

async add_many(data: list[advanced_alchemy.repository.typing.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[advanced_alchemy.repository.typing.ModelT]

async delete(item_id: Any, *, auto_commit: bool | None = None, auto_expunge: bool | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) → ModelT

async delete_many(item_ids: list[Any], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, chunk_size: int | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) → Sequence[ModelT]

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

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

async get(item_id: Any, *, auto_expunge: bool | None = None, statement: Select[tuple[ModelT]] | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) → ModelT

async get_one(*filters: StatementFilter | ColumnElement[bool], auto_expunge: bool | None = None, statement: Select[tuple[ModelT]] | 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

async get_one_or_none(*filters: StatementFilter | ColumnElement[bool], auto_expunge: bool | None = None, statement: Select[tuple[ModelT]] | 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

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]

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]

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

async update(data: ModelT, *, 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, id_attribute: str | InstrumentedAttribute[Any] | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) → ModelT

async update_many(data: list[advanced_alchemy.repository.typing.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[advanced_alchemy.repository.typing.ModelT]

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) → advanced_alchemy.repository.typing.ModelT

async upsert_many(data: list[advanced_alchemy.repository.typing.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[advanced_alchemy.repository.typing.ModelT]

async list_and_count(*filters: StatementFilter | ColumnElement[bool], auto_expunge: bool | None = None, statement: Select[tuple[ModelT]] | None = None, force_basic_query_mode: 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, order_by: list[OrderingPair] | OrderingPair | None = None, **kwargs: Any) → tuple[list[ModelT], int]

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

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

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: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None, **kwargs: 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.SQLAlchemyAsyncSlugRepositoryProtocol

Bases: SQLAlchemyAsyncRepositoryProtocol[ModelT], Protocol[ModelT]

Protocol for SQLAlchemy repositories that support slug-based operations.

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

Type Parameters:

ModelT: The SQLAlchemy model type this repository handles.

async get_by_slug(slug: str, *, 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 a model instance by its slug.

Parameters:

• slug – The slug value to search for.

• error_messages – Optional custom error message templates.

• load – Specification for eager loading of relationships.

• execution_options – Options for statement execution.

• **kwargs – Additional filtering criteria.

Returns:

The found model instance or None if not found.

Return type:

ModelT | None

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

Generate a unique slug for a given value.

Parameters:

• value_to_slugify – The string to convert to a slug.

• **kwargs – Additional parameters for slug generation.

Returns:

A unique slug derived from the input value.

Return type:

str

class advanced_alchemy.repository.SQLAlchemySyncQueryRepository

Bases: object

SQLAlchemy Query Repository.

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

__init__(*, session: 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.

error_messages: ErrorMessages | None = None

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

Get instance identified by kwargs.

Parameters:

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

• **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[RowMapping], int]

List records with total count.

Parameters:

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

• 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[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 NotFoundError if item_or_none is None.

Parameters:

item_or_none – Item to be tested for existence.

Returns:

The item, if it exists.

execute(statement: ReturningDelete[tuple[Any]] | ReturningUpdate[tuple[Any]] | Select[tuple[Any]] | Update | Delete | Select[Any]) → Result[Any]

class advanced_alchemy.repository.SQLAlchemySyncRepository

Bases: SQLAlchemySyncRepositoryProtocol[ModelT], FilterableRepository[ModelT]

Async SQLAlchemy repository implementation.

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

Type Parameters:

ModelT: The SQLAlchemy model type this repository handles.

See also

FilterableRepository

id_attribute: str = 'id'

Name of the unique identifier for the model.

loader_options: LoadSpec | None = None

Default loader options for the repository.

inherit_lazy_relationships: bool = True

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

merge_loader_options: bool = True

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

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

Default execution options for the repository.

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

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

uniquify: bool = False

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

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

__init__(*, statement: Select[tuple[ModelT]] | 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 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.

error_messages: ErrorMessages | None = None

Default error messages for the repository.

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'>) → advanced_alchemy.repository.typing.ModelT

Add data to the collection.

Parameters:

• data – Instance to be added to the collection.

• auto_expunge – Remove object from session before returning.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

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

Returns:

The added instance.

add_many(data: list[advanced_alchemy.repository.typing.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[advanced_alchemy.repository.typing.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.

• auto_commit – Commit objects before returning.

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

Returns:

The added instances.

delete(item_id: Any, *, auto_commit: bool | None = None, auto_expunge: bool | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, 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.

• auto_commit – Commit objects before returning.

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

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

• load – Set default relationships to be loaded

• execution_options – Set default execution options

Returns:

The deleted instance.

Raises:

NotFoundError – If no instance found identified by item_id.

delete_many(item_ids: list[Any], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, chunk_size: int | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, 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.

• auto_commit – Commit objects before returning.

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

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

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

• load – Set default relationships to be loaded

• execution_options – Set default execution options

Returns:

The deleted instances.

delete_where(*filters: 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.

• auto_commit – Commit objects before returning.

• 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: Any, *, auto_expunge: bool | None = None, statement: Select[tuple[ModelT]] | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, 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.

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

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

Returns:

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

get_one(*filters: StatementFilter | ColumnElement[bool], auto_expunge: bool | None = None, statement: Select[tuple[ModelT]] | 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.

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

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

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

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

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

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

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

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

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

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

• 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: ModelT, *, 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, id_attribute: str | InstrumentedAttribute[Any] | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, 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.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

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

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

• load – Set relationships to be loaded

• execution_options – Set default execution options

Returns:

The updated instance.

Raises:

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

update_many(data: list[advanced_alchemy.repository.typing.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[advanced_alchemy.repository.typing.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.

• auto_commit – Commit objects before returning.

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

• load – Set default relationships to be loaded

• execution_options – Set default execution options

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

• auto_expunge – Remove object from session before returning.

• 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) → advanced_alchemy.repository.typing.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.

• auto_refresh – Refresh object from session before returning.

• auto_commit – Commit objects before returning.

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

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

• load – Set 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[advanced_alchemy.repository.typing.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[advanced_alchemy.repository.typing.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.

• auto_commit – Commit objects before returning.

• no_merge – Skip the usage of optimized Merge statements

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

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

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

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

Bases: FilterableRepositoryProtocol[ModelT], Protocol[ModelT]

Base Protocol

id_attribute: str

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

statement: Select[tuple[ModelT]]

session: Session | scoped_session[Session]

auto_expunge: bool

auto_refresh: bool

auto_commit: bool

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

error_messages: ErrorMessages | None = None

__init__(*, statement: Select[tuple[ModelT]] | 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

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

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

static check_not_found(item_or_none: ModelT | None) → ModelT

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'>) → advanced_alchemy.repository.typing.ModelT

add_many(data: list[advanced_alchemy.repository.typing.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[advanced_alchemy.repository.typing.ModelT]

delete(item_id: Any, *, auto_commit: bool | None = None, auto_expunge: bool | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) → ModelT

delete_many(item_ids: list[Any], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, chunk_size: int | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) → Sequence[ModelT]

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

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

get(item_id: Any, *, auto_expunge: bool | None = None, statement: Select[tuple[ModelT]] | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) → ModelT

get_one(*filters: StatementFilter | ColumnElement[bool], auto_expunge: bool | None = None, statement: Select[tuple[ModelT]] | 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_one_or_none(*filters: StatementFilter | ColumnElement[bool], auto_expunge: bool | None = None, statement: Select[tuple[ModelT]] | 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_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_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]

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

update(data: ModelT, *, 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, id_attribute: str | InstrumentedAttribute[Any] | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) → ModelT

update_many(data: list[advanced_alchemy.repository.typing.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[advanced_alchemy.repository.typing.ModelT]

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) → advanced_alchemy.repository.typing.ModelT

upsert_many(data: list[advanced_alchemy.repository.typing.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[advanced_alchemy.repository.typing.ModelT]

list_and_count(*filters: StatementFilter | ColumnElement[bool], auto_expunge: bool | None = None, statement: Select[tuple[ModelT]] | None = None, force_basic_query_mode: 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, order_by: list[OrderingPair] | OrderingPair | None = None, **kwargs: Any) → tuple[list[ModelT], int]

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

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

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: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None, **kwargs: 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.SQLAlchemySyncSlugRepositoryProtocol

Bases: SQLAlchemySyncRepositoryProtocol[ModelT], Protocol[ModelT]

Protocol for SQLAlchemy repositories that support slug-based operations.

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

Type Parameters:

ModelT: The SQLAlchemy model type this repository handles.

get_by_slug(slug: str, *, 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 a model instance by its slug.

Parameters:

• slug – The slug value to search for.

• error_messages – Optional custom error message templates.

• load – Specification for eager loading of relationships.

• execution_options – Options for statement execution.

• **kwargs – Additional filtering criteria.

Returns:

The found model instance or None if not found.

Return type:

ModelT | None

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

Generate a unique slug for a given value.

Parameters:

• value_to_slugify – The string to convert to a slug.

• **kwargs – Additional parameters for slug generation.

Returns:

A unique slug derived from the input value.

Return type:

str

advanced_alchemy.repository.get_instrumented_attr(model: type[ModelProtocol], key: str | InstrumentedAttribute[Any]) → InstrumentedAttribute[Any]

Get an instrumented attribute from a model.

Parameters:

• model – SQLAlchemy model class.

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

Returns:

The instrumented attribute from the model.

Return type:

sqlalchemy.orm.InstrumentedAttribute

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

Create an ORM model instance from a dictionary of attributes.

Parameters:

• model – The SQLAlchemy model class to instantiate.

• **kwargs – Keyword arguments containing model attribute values.

Returns:

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

Return type:

ModelT