repositories

class advanced_alchemy.repository.SQLAlchemyAsyncRepository[source]

Bases: 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, **kwargs: Any) → None[source]

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.

  • **kwargs – Additional arguments.

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

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

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

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: ModelT, auto_commit: bool | None = None, auto_expunge: bool | None = None, auto_refresh: bool | None = None) → ModelT[source]

Add data to the collection.

Parameters:
Returns:

The added instance.

async add_many(data: list[ModelT], auto_commit: bool | None = None, auto_expunge: bool | None = None) → list[ModelT][source]

Add many data to the collection.

Parameters:
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) → ModelT[source]

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.

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) → list[ModelT][source]

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.

Returns:

The deleted instances.

async exists(*filters: FilterTypes | ColumnElement[bool], **kwargs: Any) → bool[source]

Return true if the object specified by kwargs exists.

Parameters:

  • *filters – Types for specific filtering operations.

  • **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 | StatementLambdaElement | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None) → ModelT[source]

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.

Returns:

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

async get_one(auto_expunge: bool | None = None, statement: Select | StatementLambdaElement | None = None, **kwargs: Any) → ModelT[source]

Get instance identified by kwargs.

Parameters:
Returns:

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

async get_one_or_none(auto_expunge: bool | None = None, statement: Select | StatementLambdaElement | None = None, **kwargs: Any) → ModelT | None[source]

Get instance identified by kwargs or None if not found.

Parameters:
Returns:

The retrieved instance or None

get_or_create(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, **kwargs: Any) → tuple[ModelT, bool][source]

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

Parameters:

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

  • **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_or_upsert(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, **kwargs: Any) → tuple[ModelT, bool][source]

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

Parameters:

  • 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

  • **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(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, **kwargs: Any) → tuple[ModelT, bool][source]

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

Parameters:

  • 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

  • **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: FilterTypes | ColumnElement[bool], statement: Select | StatementLambdaElement | None = None, **kwargs: Any) → int[source]

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

  • **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) → ModelT[source]

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.

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) → list[ModelT][source]

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

The updated instances.

Raises:

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

async list_and_count(*filters: FilterTypes | ColumnElement[bool], auto_expunge: bool | None = None, statement: Select | StatementLambdaElement | None = None, force_basic_query_mode: bool | None = None, **kwargs: Any) → tuple[list[ModelT], int][source]

List records with total count.

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

  • 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 upsert(data: ModelT, attribute_names: 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) → ModelT[source]

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.

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) → list[ModelT][source]

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 .

Returns:

The updated or created instance.

Raises:

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

async list(*filters: FilterTypes | ColumnElement[bool], auto_expunge: bool | None = None, statement: Select | StatementLambdaElement | None = None, **kwargs: Any) → list[ModelT][source]

Get a list of instances, optionally filtered.

Parameters:
Returns:

The list of instances, after filtering applied.

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

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

Perform a health check on the database.

Parameters:

session – through which we run a check statement

Returns:

True if healthy.

class advanced_alchemy.repository.SQLAlchemyAsyncQueryRepository[source]

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], **kwargs: Any) → None[source]

Repository pattern for SQLAlchemy models.

Parameters:

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

  • auto_expunge – Remove object from session before returning.

  • **kwargs – Additional arguments.

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

Get instance identified by kwargs.

Parameters:
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[source]

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

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

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

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

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

Bases: SQLAlchemyAsyncRepository[ModelT]

Extends the repository to include slug model features..

async get_by_slug(slug: str, **kwargs: Any) → ModelT | None[source]

Select record by slug value.

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

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

Bases: SQLAlchemySyncRepository[ModelT]

Extends the repository to include slug model features..

get_by_slug(slug: str, **kwargs: Any) → ModelT | None[source]

Select record by slug value.

get_available_slug(value_to_slugify: str, **kwargs: Any) → str[source]

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

Bases: 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, **kwargs: Any) → None[source]

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.

  • **kwargs – Additional arguments.

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

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

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

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: ModelT, auto_commit: bool | None = None, auto_expunge: bool | None = None, auto_refresh: bool | None = None) → ModelT[source]

Add data to the collection.

Parameters:
Returns:

The added instance.

add_many(data: list[ModelT], auto_commit: bool | None = None, auto_expunge: bool | None = None) → list[ModelT][source]

Add many data to the collection.

Parameters:
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) → ModelT[source]

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.

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) → list[ModelT][source]

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.

Returns:

The deleted instances.

exists(*filters: FilterTypes | ColumnElement[bool], **kwargs: Any) → bool[source]

Return true if the object specified by kwargs exists.

Parameters:

  • *filters – Types for specific filtering operations.

  • **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 | StatementLambdaElement | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None) → ModelT[source]

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.

Returns:

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

get_one(auto_expunge: bool | None = None, statement: Select | StatementLambdaElement | None = None, **kwargs: Any) → ModelT[source]

Get instance identified by kwargs.

Parameters:
Returns:

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

get_one_or_none(auto_expunge: bool | None = None, statement: Select | StatementLambdaElement | None = None, **kwargs: Any) → ModelT | None[source]

Get instance identified by kwargs or None if not found.

Parameters:
Returns:

The retrieved instance or None

get_or_create(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, **kwargs: Any) → tuple[ModelT, bool][source]

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

Parameters:

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

  • **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_or_upsert(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, **kwargs: Any) → tuple[ModelT, bool][source]

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

Parameters:

  • 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

  • **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(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, **kwargs: Any) → tuple[ModelT, bool][source]

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

Parameters:

  • 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

  • **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: FilterTypes | ColumnElement[bool], statement: Select | StatementLambdaElement | None = None, **kwargs: Any) → int[source]

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

  • **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) → ModelT[source]

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.

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) → list[ModelT][source]

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

The updated instances.

Raises:

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

list_and_count(*filters: FilterTypes | ColumnElement[bool], auto_expunge: bool | None = None, statement: Select | StatementLambdaElement | None = None, force_basic_query_mode: bool | None = None, **kwargs: Any) → tuple[list[ModelT], int][source]

List records with total count.

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

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

upsert(data: ModelT, attribute_names: 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) → ModelT[source]

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.

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) → list[ModelT][source]

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 .

Returns:

The updated or created instance.

Raises:

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

list(*filters: FilterTypes | ColumnElement[bool], auto_expunge: bool | None = None, statement: Select | StatementLambdaElement | None = None, **kwargs: Any) → list[ModelT][source]

Get a list of instances, optionally filtered.

Parameters:
Returns:

The list of instances, after filtering applied.

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

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

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

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], **kwargs: Any) → None[source]

Repository pattern for SQLAlchemy models.

Parameters:

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

  • auto_expunge – Remove object from session before returning.

  • **kwargs – Additional arguments.

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

Get instance identified by kwargs.

Parameters:
Returns:

The retrieved instance.

Raises:

NotFoundError – If no instance found identified by item_id.

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

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

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

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

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

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: ModelT, **kwargs: Any) → ModelT[source]

Return ORM Object from Dictionary.