repositories

final class advanced_alchemy.repository.Empty[source]

Bases: object

A sentinel class used as placeholder.

class advanced_alchemy.repository.ErrorMessages[source]

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]
not_found: str | Callable[[Exception], str]
class advanced_alchemy.repository.FilterableRepository[source]

Bases: FilterableRepositoryProtocol[ModelT]

Default implementation of a filterable repository.

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

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[tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any]] | tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any] | None = None

List or single OrderingPair to use for sorting.

class advanced_alchemy.repository.FilterableRepositoryProtocol[source]

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.

model_type: type[ModelT]

The SQLAlchemy model class this repository manages.

__init__(*args, **kwargs)
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, error_messages=None, wrap_exceptions=True, **kwargs)[source]

Repository pattern for SQLAlchemy models.

Parameters:
Return type:

None

error_messages: ErrorMessages | None = None
wrap_exceptions: bool = True
async get_one(statement, bind_group=None, **kwargs)[source]

Get instance identified by kwargs.

Parameters:

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

  • bind_group (Optional[str]) – The bind group to use for the operation.

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

  • statement (Select)

  • bind_group (str | None)

  • kwargs (Any)

Return type:

Row[Any]

Returns:

The retrieved instance.

async get_one_or_none(statement, bind_group=None, **kwargs)[source]

Get instance identified by kwargs or None if not found.

Parameters:

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

  • bind_group (Optional[str]) – The bind group to use for the operation.

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

  • statement (Select)

  • bind_group (str | None)

  • kwargs (Any)

Return type:

Optional[Row[Any]]

Returns:

The retrieved instance or None

async count(statement, bind_group=None, **kwargs)[source]

Get the count of records returned by a query.

Parameters:

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

  • bind_group (Optional[str]) – The bind group to use for the operation.

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

  • statement (Select)

  • bind_group (str | None)

  • kwargs (Any)

Return type:

int

Returns:

Count of records returned by query, ignoring pagination.

async list_and_count(statement, count_with_window_function=None, bind_group=None, **kwargs)[source]

List records with total count.

Parameters:

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

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

  • bind_group (Optional[str]) – The bind group to use for the operation.

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

  • statement (Select)

  • count_with_window_function (bool | None)

  • bind_group (str | None)

  • kwargs (Any)

Return type:

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

Returns:

Count of records returned by query, ignoring pagination.

async list(statement, bind_group=None, **kwargs)[source]

Get a list of instances, optionally filtered.

Parameters:

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

  • bind_group (Optional[str]) – The bind group to use for the operation.

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

  • statement (Select)

  • bind_group (str | None)

  • kwargs (Any)

Return type:

list[Row[Any]]

Returns:

The list of instances, after filtering applied.

static check_not_found(item_or_none)[source]

Raise NotFoundError if item_or_none is None.

Parameters:

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

  • item_or_none (T | None)

Raises:

NotFoundError – If item_or_none is None

Return type:

TypeVar(T)

Returns:

The item, if it exists.

async execute(statement, execution_options=None)[source]
Return type:

Result[Any]

Parameters:
class advanced_alchemy.repository.SQLAlchemyAsyncRepository[source]

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: Sequence[_AbstractLoad | Literal['*'] | InstrumentedAttribute[Any] | RelationshipProperty[Any] | MapperProperty[Any] | Sequence[_AbstractLoad | Literal['*'] | InstrumentedAttribute[Any] | RelationshipProperty[Any] | MapperProperty[Any]]] | _AbstractLoad | Literal['*'] | InstrumentedAttribute[Any] | RelationshipProperty[Any] | MapperProperty[Any] | ExecutableOption | Sequence[ExecutableOption] | None = None

Default loader options for the repository.

inherit_lazy_relationships: bool = True

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

merge_loader_options: bool = True

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

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

Default execution options for the repository.

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

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

uniquify: bool = False

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

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

__init__(*, statement=None, session, auto_expunge=False, auto_refresh=True, auto_commit=False, order_by=None, error_messages=Empty, load=None, execution_options=None, wrap_exceptions=True, uniquify=None, count_with_window_function=None, cache_manager=None, bind_group=None, **kwargs)[source]

Repository for SQLAlchemy models.

Parameters:
Return type:

None

error_messages: ErrorMessages | None = None

Default error messages for the repository.

wrap_exceptions: bool = True

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

count_with_window_function: bool = True

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

classmethod get_id_attribute_value(item, id_attribute=None)[source]

Get value of attribute named as id_attribute on item.

Parameters:

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

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

  • item (ModelT | type[ModelT])

  • id_attribute (str | InstrumentedAttribute[Any] | None)

Return type:

Any

Returns:

The value of attribute on item named as id_attribute.

classmethod set_id_attribute_value(item_id, item, id_attribute=None)[source]

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

Parameters:

  • item_id (Any) – Value of ID to be set on instance

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

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

  • item_id (Any)

  • item (ModelT)

  • id_attribute (str | InstrumentedAttribute[Any] | None)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

Item with item_id set to id_attribute

static check_not_found(item_or_none)[source]

Raise advanced_alchemy.exceptions.NotFoundError if item_or_none is None.

Parameters:

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

  • item_or_none (ModelT | None)

Raises:

NotFoundError – If item_or_none is None

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

The item, if it exists.

async add(data, *, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=Empty, bind_group=None)[source]

Add data to the collection.

Parameters:

  • data (TypeVar(ModelT, bound= base.ModelProtocol)) – Instance to be added to the collection.

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

  • auto_refresh (Optional[bool]) – Refresh object from session before returning.

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • bind_group (Optional[str]) – Optional routing group for multi-master configurations.

  • data (ModelT)

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • auto_refresh (bool | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • bind_group (str | None)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

The added instance.

async add_many(data, *, auto_commit=None, auto_expunge=None, error_messages=Empty, bind_group=None)[source]

Add many data to the collection.

Parameters:

  • data (list[TypeVar(ModelT, bound= base.ModelProtocol)]) – list of Instances to be added to the collection.

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • bind_group (Optional[str]) – Optional routing group for multi-master configurations.

  • data (list[ModelT])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • bind_group (str | None)

Return type:

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

Returns:

The added instances.

async delete(item_id, *, auto_commit=None, auto_expunge=None, id_attribute=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None)[source]

Delete instance identified by item_id.

Parameters:

  • item_id (Any) – Identifier of instance to be deleted.

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

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

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • bind_group (Optional[str]) – Optional routing group for multi-master configurations.

  • item_id (Any)

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • id_attribute (str | InstrumentedAttribute[Any] | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • bind_group (str | None)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

The deleted instance.

async delete_many(item_ids, *, auto_commit=None, auto_expunge=None, id_attribute=None, chunk_size=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None)[source]

Delete instance identified by item_id.

Parameters:

  • item_ids (list[Any]) – Identifier of instance to be deleted.

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

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

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

  • chunk_size (Optional[int]) – Allows customization of the insertmanyvalues_max_parameters setting for the driver. Defaults to 950 if left unset.

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • bind_group (Optional[str]) – Optional routing group for multi-master configurations.

  • item_ids (list[Any])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • id_attribute (str | InstrumentedAttribute[Any] | None)

  • chunk_size (int | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • bind_group (str | None)

Return type:

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

Returns:

The deleted instances.

async delete_where(*filters, auto_commit=None, auto_expunge=None, error_messages=Empty, sanity_check=True, load=None, execution_options=None, uniquify=None, bind_group=None, **kwargs)[source]

Delete instances specified by referenced kwargs and filters.

Parameters:
Raises:

RepositoryError – If the number of deleted rows does not match the number of selected instances

Return type:

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

Returns:

The deleted instances.

async exists(*filters, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None, **kwargs)[source]

Return true if the object specified by kwargs exists.

Parameters:
Return type:

bool

Returns:

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

async get(item_id, *, auto_expunge=None, statement=None, id_attribute=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, with_for_update=None, use_cache=True, bind_group=None)[source]

Get instance identified by item_id.

Parameters:

  • item_id (Any) – Identifier of the instance to be retrieved.

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

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

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • with_for_update (Union[sqlalchemy.sql.selectable.ForUpdateArg, None, bool, Dict[str, Any]]) – Optional FOR UPDATE clause / parameters to apply to the SELECT statement.

  • use_cache (bool) – Whether to use caching for this query (default True).

  • bind_group (Optional[str]) – Optional routing group to use for the operation.

  • item_id (Any)

  • auto_expunge (bool | None)

  • statement (Select | None)

  • id_attribute (str | InstrumentedAttribute[Any] | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • use_cache (bool)

  • bind_group (str | None)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

The retrieved instance.

async get_one(*filters, auto_expunge=None, statement=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, with_for_update=None, bind_group=None, **kwargs)[source]

Get instance identified by kwargs.

Parameters:

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

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • with_for_update (Union[sqlalchemy.sql.selectable.ForUpdateArg, None, bool, Dict[str, Any]]) – Optional FOR UPDATE clause / parameters to apply to the SELECT statement.

  • bind_group (Optional[str]) – Optional routing group to use for the operation.

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

  • filters (StatementFilter | ColumnElement[bool])

  • auto_expunge (bool | None)

  • statement (Select | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • bind_group (str | None)

  • kwargs (Any)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

The retrieved instance.

async get_one_or_none(*filters, auto_expunge=None, statement=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, with_for_update=None, bind_group=None, **kwargs)[source]

Get instance identified by kwargs or None if not found.

Parameters:

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

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • with_for_update (Union[sqlalchemy.sql.selectable.ForUpdateArg, None, bool, Dict[str, Any]]) – Optional FOR UPDATE clause / parameters to apply to the SELECT statement.

  • bind_group (Optional[str]) – Optional routing group to use for the operation.

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

  • filters (StatementFilter | ColumnElement[bool])

  • auto_expunge (bool | None)

  • statement (Select | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • bind_group (str | None)

  • kwargs (Any)

Return type:

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

Returns:

The retrieved instance or None

async get_or_upsert(*filters, match_fields=None, upsert=True, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None, **kwargs)[source]

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

Parameters:

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

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

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

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

  • with_for_update (Union[sqlalchemy.sql.selectable.ForUpdateArg, None, bool, Dict[str, Any]]) – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

  • auto_refresh (Optional[bool]) – Refresh object from session before returning.

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • bind_group (Optional[str]) – Optional routing group for multi-master configurations.

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

  • filters (StatementFilter | ColumnElement[bool])

  • match_fields (list[str] | str | None)

  • upsert (bool)

  • attribute_names (Iterable[str] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • auto_refresh (bool | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • bind_group (str | None)

  • kwargs (Any)

Return type:

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

Returns:

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

async get_and_update(*filters, match_fields=None, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None, **kwargs)[source]

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

Parameters:

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

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

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

  • with_for_update (Union[sqlalchemy.sql.selectable.ForUpdateArg, None, bool, Dict[str, Any]]) – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

  • auto_refresh (Optional[bool]) – Refresh object from session before returning.

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • bind_group (Optional[str]) – Optional routing group for multi-master configurations.

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

  • filters (StatementFilter | ColumnElement[bool])

  • match_fields (list[str] | str | None)

  • attribute_names (Iterable[str] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • auto_refresh (bool | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • bind_group (str | None)

  • kwargs (Any)

Return type:

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

Returns:

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

async count(*filters, statement=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None, **kwargs)[source]

Get the count of records returned by a query.

Parameters:
Return type:

int

Returns:

Count of records returned by query, ignoring pagination.

async update(data, *, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, id_attribute=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None)[source]

Update instance with the attribute values present on data.

Parameters:

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

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

  • with_for_update (Union[sqlalchemy.sql.selectable.ForUpdateArg, None, bool, Dict[str, Any]]) – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

  • auto_refresh (Optional[bool]) – Refresh object from session before returning.

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

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • bind_group (Optional[str]) – Optional routing group for multi-master configurations.

  • data (ModelT)

  • attribute_names (Iterable[str] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • auto_refresh (bool | None)

  • id_attribute (str | InstrumentedAttribute[Any] | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • bind_group (str | None)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

The updated instance.

async update_many(data, *, auto_commit=None, auto_expunge=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None)[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:

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

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • bind_group (Optional[str]) – Optional routing group for multi-master configurations.

  • data (list[ModelT])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • bind_group (str | None)

Return type:

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

Returns:

The updated instances.

async list_and_count(*filters, statement=None, auto_expunge=None, count_with_window_function=None, order_by=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, use_cache=True, bind_group=None, **kwargs)[source]

List records with total count.

Parameters:
Return type:

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

Returns:

Count of records returned by query, ignoring pagination.

async upsert(data, *, attribute_names=None, with_for_update=None, auto_expunge=None, auto_commit=None, auto_refresh=None, match_fields=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None)[source]

Modify or create instance.

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

Parameters:

  • data (TypeVar(ModelT, bound= base.ModelProtocol)) – Instance to update existing, or be created. Identifier used to determine if an existing instance exists is the value of an attribute on data named as value of self.id_attribute.

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

  • with_for_update (Union[sqlalchemy.sql.selectable.ForUpdateArg, None, bool, Dict[str, Any]]) – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

  • auto_refresh (Optional[bool]) – Refresh object from session before returning.

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

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • bind_group (Optional[str]) – Optional routing group for multi-master configurations.

  • data (ModelT)

  • attribute_names (Iterable[str] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • auto_expunge (bool | None)

  • auto_commit (bool | None)

  • auto_refresh (bool | None)

  • match_fields (list[str] | str | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • bind_group (str | None)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

The updated or created instance.

async upsert_many(data, *, auto_expunge=None, auto_commit=None, no_merge=False, match_fields=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None)[source]

Modify or create multiple instances.

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

!!! tip

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

Parameters:

  • data (list[TypeVar(ModelT, bound= base.ModelProtocol)]) – Instance to update existing, or be created. Identifier used to determine if an existing instance exists is the value of an attribute on data named as value of id_attribute.

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

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

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

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • bind_group (Optional[str]) – Optional routing group to use for the operation.

  • data (list[ModelT])

  • auto_expunge (bool | None)

  • auto_commit (bool | None)

  • no_merge (bool)

  • match_fields (list[str] | str | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • bind_group (str | None)

Return type:

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

Returns:

The updated or created instance.

async list(*filters, auto_expunge=None, statement=None, order_by=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, use_cache=True, bind_group=None, **kwargs)[source]

Get a list of instances, optionally filtered.

Parameters:
Return type:

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

Returns:

The list of instances, after filtering applied.

async classmethod check_health(session)[source]

Perform a health check on the database.

Parameters:
Return type:

bool

Returns:

True if healthy.

class advanced_alchemy.repository.SQLAlchemyAsyncRepositoryProtocol[source]

Bases: FilterableRepositoryProtocol[ModelT], Protocol[ModelT]

Base Protocol

id_attribute: str
match_fields: list[str] | str | None = None
statement: Select
session: AsyncSession | async_scoped_session[AsyncSession]
auto_expunge: bool
auto_refresh: bool
auto_commit: bool
order_by: list[tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any]] | tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any] | None = None
error_messages: ErrorMessages | None = None
wrap_exceptions: bool = True
__init__(*, statement=None, session, auto_expunge=False, auto_refresh=True, auto_commit=False, load=None, execution_options=None, order_by=None, error_messages=Empty, wrap_exceptions=True, **kwargs)[source]
Parameters:

  • statement (Select | None)

  • session (AsyncSession | async_scoped_session[AsyncSession])

  • auto_expunge (bool)

  • auto_refresh (bool)

  • auto_commit (bool)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • order_by (list[tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any]] | tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any] | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • wrap_exceptions (bool)

  • kwargs (Any)

Return type:

None

classmethod get_id_attribute_value(item, id_attribute=None)[source]
Return type:

Any

Parameters:
classmethod set_id_attribute_value(item_id, item, id_attribute=None)[source]
Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Parameters:
static check_not_found(item_or_none)[source]
Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Parameters:

item_or_none (ModelT | None)

async add(data, *, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=Empty, bind_group=None)[source]
Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Parameters:
async add_many(data, *, auto_commit=None, auto_expunge=None, error_messages=Empty, bind_group=None)[source]
Return type:

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

Parameters:
async delete(item_id, *, auto_commit=None, auto_expunge=None, id_attribute=None, error_messages=Empty, load=None, execution_options=None, bind_group=None)[source]
Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Parameters:

  • item_id (Any)

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • id_attribute (str | InstrumentedAttribute[Any] | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

async delete_many(item_ids, *, auto_commit=None, auto_expunge=None, id_attribute=None, chunk_size=None, error_messages=Empty, load=None, execution_options=None, bind_group=None)[source]
Return type:

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

Parameters:

  • item_ids (list[Any])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • id_attribute (str | InstrumentedAttribute[Any] | None)

  • chunk_size (int | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

async delete_where(*filters, auto_commit=None, auto_expunge=None, load=None, error_messages=Empty, execution_options=None, sanity_check=True, bind_group=None, **kwargs)[source]
Return type:

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

Parameters:

  • filters (StatementFilter | ColumnElement[bool])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • execution_options (dict[str, Any] | None)

  • sanity_check (bool)

  • bind_group (str | None)

  • kwargs (Any)

async exists(*filters, load=None, error_messages=Empty, execution_options=None, bind_group=None, **kwargs)[source]
Return type:

bool

Parameters:

  • filters (StatementFilter | ColumnElement[bool])

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

  • kwargs (Any)

async get(item_id, *, auto_expunge=None, statement=None, id_attribute=None, error_messages=Empty, load=None, execution_options=None, with_for_update=None, bind_group=None)[source]
Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Parameters:

  • item_id (Any)

  • auto_expunge (bool | None)

  • statement (Select | None)

  • id_attribute (str | InstrumentedAttribute[Any] | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • bind_group (str | None)

async get_one(*filters, auto_expunge=None, statement=None, error_messages=Empty, load=None, execution_options=None, with_for_update=None, bind_group=None, **kwargs)[source]
Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Parameters:

  • filters (StatementFilter | ColumnElement[bool])

  • auto_expunge (bool | None)

  • statement (Select | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • bind_group (str | None)

  • kwargs (Any)

async get_one_or_none(*filters, auto_expunge=None, statement=None, error_messages=Empty, load=None, execution_options=None, with_for_update=None, bind_group=None, **kwargs)[source]
Return type:

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

Parameters:

  • filters (StatementFilter | ColumnElement[bool])

  • auto_expunge (bool | None)

  • statement (Select | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • bind_group (str | None)

  • kwargs (Any)

async get_or_upsert(*filters, match_fields=None, upsert=True, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=Empty, load=None, execution_options=None, bind_group=None, **kwargs)[source]
Return type:

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

Parameters:

  • filters (StatementFilter | ColumnElement[bool])

  • match_fields (list[str] | str | None)

  • upsert (bool)

  • attribute_names (Iterable[str] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • auto_refresh (bool | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

  • kwargs (Any)

async get_and_update(*filters, match_fields=None, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=Empty, load=None, execution_options=None, bind_group=None, **kwargs)[source]
Return type:

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

Parameters:

  • filters (StatementFilter | ColumnElement[bool])

  • match_fields (list[str] | str | None)

  • attribute_names (Iterable[str] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • auto_refresh (bool | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

  • kwargs (Any)

async count(*filters, statement=None, load=None, error_messages=Empty, execution_options=None, bind_group=None, **kwargs)[source]
Return type:

int

Parameters:

  • filters (StatementFilter | ColumnElement[bool])

  • statement (Select | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

  • kwargs (Any)

async update(data, *, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, id_attribute=None, error_messages=Empty, load=None, execution_options=None, bind_group=None)[source]
Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Parameters:

  • data (ModelT)

  • attribute_names (Iterable[str] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • auto_refresh (bool | None)

  • id_attribute (str | InstrumentedAttribute[Any] | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

async update_many(data, *, auto_commit=None, auto_expunge=None, error_messages=Empty, load=None, execution_options=None, bind_group=None)[source]
Return type:

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

Parameters:

  • data (list[ModelT])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

async upsert(data, *, attribute_names=None, with_for_update=None, auto_expunge=None, auto_commit=None, auto_refresh=None, match_fields=None, error_messages=Empty, load=None, execution_options=None, bind_group=None)[source]
Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Parameters:

  • data (ModelT)

  • attribute_names (Iterable[str] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • auto_expunge (bool | None)

  • auto_commit (bool | None)

  • auto_refresh (bool | None)

  • match_fields (list[str] | str | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

async upsert_many(data, *, auto_expunge=None, auto_commit=None, no_merge=False, match_fields=None, error_messages=Empty, load=None, execution_options=None, bind_group=None)[source]
Return type:

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

Parameters:

  • data (list[ModelT])

  • auto_expunge (bool | None)

  • auto_commit (bool | None)

  • no_merge (bool)

  • match_fields (list[str] | str | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

async list_and_count(*filters, auto_expunge=None, statement=None, count_with_window_function=None, error_messages=Empty, load=None, execution_options=None, order_by=None, use_cache=True, bind_group=None, **kwargs)[source]
Return type:

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

Parameters:

  • filters (StatementFilter | ColumnElement[bool])

  • auto_expunge (bool | None)

  • statement (Select | None)

  • count_with_window_function (bool | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • order_by (list[tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any]] | tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any] | None)

  • use_cache (bool)

  • bind_group (str | None)

  • kwargs (Any)

async list(*filters, auto_expunge=None, statement=None, error_messages=Empty, load=None, execution_options=None, order_by=None, use_cache=True, bind_group=None, **kwargs)[source]
Return type:

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

Parameters:

  • filters (StatementFilter | ColumnElement[bool])

  • auto_expunge (bool | None)

  • statement (Select | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • order_by (list[tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any]] | tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any] | None)

  • use_cache (bool)

  • bind_group (str | None)

  • kwargs (Any)

async classmethod check_health(session)[source]
Return type:

bool

Parameters:

session (AsyncSession | async_scoped_session[AsyncSession])

class advanced_alchemy.repository.SQLAlchemyAsyncSlugRepository[source]

Bases: SQLAlchemyAsyncRepository[ModelT], SQLAlchemyAsyncSlugRepositoryProtocol[ModelT]

Extends the repository to include slug model features..

async get_by_slug(slug, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None, **kwargs)[source]

Select record by slug value.

Return type:

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

Returns:

The model instance or None if not found.

Parameters:

  • slug (str)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • bind_group (str | None)

  • kwargs (Any)

async get_available_slug(value_to_slugify, **kwargs)[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 (Any) – stuff

  • value_to_slugify (str)

  • kwargs (Any)

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

Bases: SQLAlchemyAsyncRepositoryProtocol[ModelT], Protocol[ModelT]

Protocol for SQLAlchemy repositories that support slug-based operations.

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

Type Parameters:

ModelT: The SQLAlchemy model type this repository handles.

async get_by_slug(slug, *, error_messages=Empty, load=None, execution_options=None, bind_group=None, **kwargs)[source]

Get a model instance by its slug.

Parameters:

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – Optional custom error message templates.

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Specification for eager loading of relationships.

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

  • bind_group (Optional[str]) – Optional routing group to use for the operation.

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

  • slug (str)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

  • kwargs (Any)

Returns:

The found model instance or None if not found.

Return type:

ModelT | None

async get_available_slug(value_to_slugify, **kwargs)[source]

Generate a unique slug for a given value.

Parameters:

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

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

  • value_to_slugify (str)

  • kwargs (Any)

Returns:

A unique slug derived from the input value.

Return type:

str

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, error_messages=None, wrap_exceptions=True, **kwargs)[source]

Repository pattern for SQLAlchemy models.

Parameters:
Return type:

None

error_messages: ErrorMessages | None = None
wrap_exceptions: bool = True
get_one(statement, bind_group=None, **kwargs)[source]

Get instance identified by kwargs.

Parameters:

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

  • bind_group (Optional[str]) – The bind group to use for the operation.

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

  • statement (Select)

  • bind_group (str | None)

  • kwargs (Any)

Return type:

Row[Any]

Returns:

The retrieved instance.

get_one_or_none(statement, bind_group=None, **kwargs)[source]

Get instance identified by kwargs or None if not found.

Parameters:

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

  • bind_group (Optional[str]) – The bind group to use for the operation.

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

  • statement (Select)

  • bind_group (str | None)

  • kwargs (Any)

Return type:

Optional[Row[Any]]

Returns:

The retrieved instance or None

count(statement, bind_group=None, **kwargs)[source]

Get the count of records returned by a query.

Parameters:

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

  • bind_group (Optional[str]) – The bind group to use for the operation.

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

  • statement (Select)

  • bind_group (str | None)

  • kwargs (Any)

Return type:

int

Returns:

Count of records returned by query, ignoring pagination.

list_and_count(statement, count_with_window_function=None, bind_group=None, **kwargs)[source]

List records with total count.

Parameters:

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

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

  • bind_group (Optional[str]) – The bind group to use for the operation.

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

  • statement (Select)

  • count_with_window_function (bool | None)

  • bind_group (str | None)

  • kwargs (Any)

Return type:

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

Returns:

Count of records returned by query, ignoring pagination.

list(statement, bind_group=None, **kwargs)[source]

Get a list of instances, optionally filtered.

Parameters:

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

  • bind_group (Optional[str]) – The bind group to use for the operation.

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

  • statement (Select)

  • bind_group (str | None)

  • kwargs (Any)

Return type:

list[Row[Any]]

Returns:

The list of instances, after filtering applied.

static check_not_found(item_or_none)[source]

Raise NotFoundError if item_or_none is None.

Parameters:

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

  • item_or_none (T | None)

Raises:

NotFoundError – If item_or_none is None

Return type:

TypeVar(T)

Returns:

The item, if it exists.

execute(statement, execution_options=None)[source]
Return type:

Result[Any]

Parameters:
class advanced_alchemy.repository.SQLAlchemySyncRepository[source]

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: Sequence[_AbstractLoad | Literal['*'] | InstrumentedAttribute[Any] | RelationshipProperty[Any] | MapperProperty[Any] | Sequence[_AbstractLoad | Literal['*'] | InstrumentedAttribute[Any] | RelationshipProperty[Any] | MapperProperty[Any]]] | _AbstractLoad | Literal['*'] | InstrumentedAttribute[Any] | RelationshipProperty[Any] | MapperProperty[Any] | ExecutableOption | Sequence[ExecutableOption] | None = None

Default loader options for the repository.

inherit_lazy_relationships: bool = True

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

merge_loader_options: bool = True

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

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

Default execution options for the repository.

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

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

uniquify: bool = False

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

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

__init__(*, statement=None, session, auto_expunge=False, auto_refresh=True, auto_commit=False, order_by=None, error_messages=Empty, load=None, execution_options=None, wrap_exceptions=True, uniquify=None, count_with_window_function=None, cache_manager=None, bind_group=None, **kwargs)[source]

Repository for SQLAlchemy models.

Parameters:
Return type:

None

error_messages: ErrorMessages | None = None

Default error messages for the repository.

wrap_exceptions: bool = True

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

count_with_window_function: bool = True

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

classmethod get_id_attribute_value(item, id_attribute=None)[source]

Get value of attribute named as id_attribute on item.

Parameters:

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

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

  • item (ModelT | type[ModelT])

  • id_attribute (str | InstrumentedAttribute[Any] | None)

Return type:

Any

Returns:

The value of attribute on item named as id_attribute.

classmethod set_id_attribute_value(item_id, item, id_attribute=None)[source]

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

Parameters:

  • item_id (Any) – Value of ID to be set on instance

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

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

  • item_id (Any)

  • item (ModelT)

  • id_attribute (str | InstrumentedAttribute[Any] | None)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

Item with item_id set to id_attribute

static check_not_found(item_or_none)[source]

Raise advanced_alchemy.exceptions.NotFoundError if item_or_none is None.

Parameters:

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

  • item_or_none (ModelT | None)

Raises:

NotFoundError – If item_or_none is None

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

The item, if it exists.

add(data, *, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=Empty, bind_group=None)[source]

Add data to the collection.

Parameters:

  • data (TypeVar(ModelT, bound= base.ModelProtocol)) – Instance to be added to the collection.

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

  • auto_refresh (Optional[bool]) – Refresh object from session before returning.

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • bind_group (Optional[str]) – Optional routing group for multi-master configurations.

  • data (ModelT)

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • auto_refresh (bool | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • bind_group (str | None)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

The added instance.

add_many(data, *, auto_commit=None, auto_expunge=None, error_messages=Empty, bind_group=None)[source]

Add many data to the collection.

Parameters:

  • data (list[TypeVar(ModelT, bound= base.ModelProtocol)]) – list of Instances to be added to the collection.

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • bind_group (Optional[str]) – Optional routing group for multi-master configurations.

  • data (list[ModelT])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • bind_group (str | None)

Return type:

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

Returns:

The added instances.

delete(item_id, *, auto_commit=None, auto_expunge=None, id_attribute=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None)[source]

Delete instance identified by item_id.

Parameters:

  • item_id (Any) – Identifier of instance to be deleted.

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

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

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • bind_group (Optional[str]) – Optional routing group for multi-master configurations.

  • item_id (Any)

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • id_attribute (str | InstrumentedAttribute[Any] | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • bind_group (str | None)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

The deleted instance.

delete_many(item_ids, *, auto_commit=None, auto_expunge=None, id_attribute=None, chunk_size=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None)[source]

Delete instance identified by item_id.

Parameters:

  • item_ids (list[Any]) – Identifier of instance to be deleted.

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

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

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

  • chunk_size (Optional[int]) – Allows customization of the insertmanyvalues_max_parameters setting for the driver. Defaults to 950 if left unset.

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • bind_group (Optional[str]) – Optional routing group for multi-master configurations.

  • item_ids (list[Any])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • id_attribute (str | InstrumentedAttribute[Any] | None)

  • chunk_size (int | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • bind_group (str | None)

Return type:

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

Returns:

The deleted instances.

delete_where(*filters, auto_commit=None, auto_expunge=None, error_messages=Empty, sanity_check=True, load=None, execution_options=None, uniquify=None, bind_group=None, **kwargs)[source]

Delete instances specified by referenced kwargs and filters.

Parameters:
Raises:

RepositoryError – If the number of deleted rows does not match the number of selected instances

Return type:

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

Returns:

The deleted instances.

exists(*filters, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None, **kwargs)[source]

Return true if the object specified by kwargs exists.

Parameters:
Return type:

bool

Returns:

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

get(item_id, *, auto_expunge=None, statement=None, id_attribute=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, with_for_update=None, use_cache=True, bind_group=None)[source]

Get instance identified by item_id.

Parameters:

  • item_id (Any) – Identifier of the instance to be retrieved.

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

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

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • with_for_update (Union[sqlalchemy.sql.selectable.ForUpdateArg, None, bool, Dict[str, Any]]) – Optional FOR UPDATE clause / parameters to apply to the SELECT statement.

  • use_cache (bool) – Whether to use caching for this query (default True).

  • bind_group (Optional[str]) – Optional routing group to use for the operation.

  • item_id (Any)

  • auto_expunge (bool | None)

  • statement (Select | None)

  • id_attribute (str | InstrumentedAttribute[Any] | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • use_cache (bool)

  • bind_group (str | None)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

The retrieved instance.

get_one(*filters, auto_expunge=None, statement=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, with_for_update=None, bind_group=None, **kwargs)[source]

Get instance identified by kwargs.

Parameters:

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

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • with_for_update (Union[sqlalchemy.sql.selectable.ForUpdateArg, None, bool, Dict[str, Any]]) – Optional FOR UPDATE clause / parameters to apply to the SELECT statement.

  • bind_group (Optional[str]) – Optional routing group to use for the operation.

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

  • filters (StatementFilter | ColumnElement[bool])

  • auto_expunge (bool | None)

  • statement (Select | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • bind_group (str | None)

  • kwargs (Any)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

The retrieved instance.

get_one_or_none(*filters, auto_expunge=None, statement=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, with_for_update=None, bind_group=None, **kwargs)[source]

Get instance identified by kwargs or None if not found.

Parameters:

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

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • with_for_update (Union[sqlalchemy.sql.selectable.ForUpdateArg, None, bool, Dict[str, Any]]) – Optional FOR UPDATE clause / parameters to apply to the SELECT statement.

  • bind_group (Optional[str]) – Optional routing group to use for the operation.

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

  • filters (StatementFilter | ColumnElement[bool])

  • auto_expunge (bool | None)

  • statement (Select | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • bind_group (str | None)

  • kwargs (Any)

Return type:

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

Returns:

The retrieved instance or None

get_or_upsert(*filters, match_fields=None, upsert=True, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None, **kwargs)[source]

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

Parameters:

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

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

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

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

  • with_for_update (Union[sqlalchemy.sql.selectable.ForUpdateArg, None, bool, Dict[str, Any]]) – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

  • auto_refresh (Optional[bool]) – Refresh object from session before returning.

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • bind_group (Optional[str]) – Optional routing group for multi-master configurations.

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

  • filters (StatementFilter | ColumnElement[bool])

  • match_fields (list[str] | str | None)

  • upsert (bool)

  • attribute_names (Iterable[str] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • auto_refresh (bool | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • bind_group (str | None)

  • kwargs (Any)

Return type:

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

Returns:

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

get_and_update(*filters, match_fields=None, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None, **kwargs)[source]

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

Parameters:

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

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

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

  • with_for_update (Union[sqlalchemy.sql.selectable.ForUpdateArg, None, bool, Dict[str, Any]]) – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

  • auto_refresh (Optional[bool]) – Refresh object from session before returning.

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • bind_group (Optional[str]) – Optional routing group for multi-master configurations.

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

  • filters (StatementFilter | ColumnElement[bool])

  • match_fields (list[str] | str | None)

  • attribute_names (Iterable[str] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • auto_refresh (bool | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • bind_group (str | None)

  • kwargs (Any)

Return type:

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

Returns:

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

count(*filters, statement=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None, **kwargs)[source]

Get the count of records returned by a query.

Parameters:
Return type:

int

Returns:

Count of records returned by query, ignoring pagination.

update(data, *, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, id_attribute=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None)[source]

Update instance with the attribute values present on data.

Parameters:

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

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

  • with_for_update (Union[sqlalchemy.sql.selectable.ForUpdateArg, None, bool, Dict[str, Any]]) – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

  • auto_refresh (Optional[bool]) – Refresh object from session before returning.

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

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • bind_group (Optional[str]) – Optional routing group for multi-master configurations.

  • data (ModelT)

  • attribute_names (Iterable[str] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • auto_refresh (bool | None)

  • id_attribute (str | InstrumentedAttribute[Any] | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • bind_group (str | None)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

The updated instance.

update_many(data, *, auto_commit=None, auto_expunge=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None)[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:

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

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • bind_group (Optional[str]) – Optional routing group for multi-master configurations.

  • data (list[ModelT])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • bind_group (str | None)

Return type:

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

Returns:

The updated instances.

list_and_count(*filters, statement=None, auto_expunge=None, count_with_window_function=None, order_by=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, use_cache=True, bind_group=None, **kwargs)[source]

List records with total count.

Parameters:
Return type:

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

Returns:

Count of records returned by query, ignoring pagination.

upsert(data, *, attribute_names=None, with_for_update=None, auto_expunge=None, auto_commit=None, auto_refresh=None, match_fields=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None)[source]

Modify or create instance.

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

Parameters:

  • data (TypeVar(ModelT, bound= base.ModelProtocol)) – Instance to update existing, or be created. Identifier used to determine if an existing instance exists is the value of an attribute on data named as value of self.id_attribute.

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

  • with_for_update (Union[sqlalchemy.sql.selectable.ForUpdateArg, None, bool, Dict[str, Any]]) – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

  • auto_refresh (Optional[bool]) – Refresh object from session before returning.

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

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • bind_group (Optional[str]) – Optional routing group for multi-master configurations.

  • data (ModelT)

  • attribute_names (Iterable[str] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • auto_expunge (bool | None)

  • auto_commit (bool | None)

  • auto_refresh (bool | None)

  • match_fields (list[str] | str | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • bind_group (str | None)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

The updated or created instance.

upsert_many(data, *, auto_expunge=None, auto_commit=None, no_merge=False, match_fields=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None)[source]

Modify or create multiple instances.

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

!!! tip

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

Parameters:

  • data (list[TypeVar(ModelT, bound= base.ModelProtocol)]) – Instance to update existing, or be created. Identifier used to determine if an existing instance exists is the value of an attribute on data named as value of id_attribute.

  • auto_expunge (Optional[bool]) – Remove object from session before returning.

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

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

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded

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

  • uniquify (Optional[bool]) – Optionally apply the unique() method to results before returning.

  • bind_group (Optional[str]) – Optional routing group to use for the operation.

  • data (list[ModelT])

  • auto_expunge (bool | None)

  • auto_commit (bool | None)

  • no_merge (bool)

  • match_fields (list[str] | str | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • bind_group (str | None)

Return type:

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

Returns:

The updated or created instance.

list(*filters, auto_expunge=None, statement=None, order_by=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, use_cache=True, bind_group=None, **kwargs)[source]

Get a list of instances, optionally filtered.

Parameters:
Return type:

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

Returns:

The list of instances, after filtering applied.

classmethod check_health(session)[source]

Perform a health check on the database.

Parameters:
Return type:

bool

Returns:

True if healthy.

class advanced_alchemy.repository.SQLAlchemySyncRepositoryProtocol[source]

Bases: FilterableRepositoryProtocol[ModelT], Protocol[ModelT]

Base Protocol

id_attribute: str
match_fields: list[str] | str | None = None
statement: Select
session: Session | scoped_session[Session]
auto_expunge: bool
auto_refresh: bool
auto_commit: bool
order_by: list[tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any]] | tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any] | None = None
error_messages: ErrorMessages | None = None
wrap_exceptions: bool = True
__init__(*, statement=None, session, auto_expunge=False, auto_refresh=True, auto_commit=False, load=None, execution_options=None, order_by=None, error_messages=Empty, wrap_exceptions=True, **kwargs)[source]
Parameters:

  • statement (Select | None)

  • session (Session | scoped_session[Session])

  • auto_expunge (bool)

  • auto_refresh (bool)

  • auto_commit (bool)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • order_by (list[tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any]] | tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any] | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • wrap_exceptions (bool)

  • kwargs (Any)

Return type:

None

classmethod get_id_attribute_value(item, id_attribute=None)[source]
Return type:

Any

Parameters:
classmethod set_id_attribute_value(item_id, item, id_attribute=None)[source]
Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Parameters:
static check_not_found(item_or_none)[source]
Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Parameters:

item_or_none (ModelT | None)

add(data, *, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=Empty, bind_group=None)[source]
Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Parameters:
add_many(data, *, auto_commit=None, auto_expunge=None, error_messages=Empty, bind_group=None)[source]
Return type:

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

Parameters:
delete(item_id, *, auto_commit=None, auto_expunge=None, id_attribute=None, error_messages=Empty, load=None, execution_options=None, bind_group=None)[source]
Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Parameters:

  • item_id (Any)

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • id_attribute (str | InstrumentedAttribute[Any] | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

delete_many(item_ids, *, auto_commit=None, auto_expunge=None, id_attribute=None, chunk_size=None, error_messages=Empty, load=None, execution_options=None, bind_group=None)[source]
Return type:

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

Parameters:

  • item_ids (list[Any])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • id_attribute (str | InstrumentedAttribute[Any] | None)

  • chunk_size (int | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

delete_where(*filters, auto_commit=None, auto_expunge=None, load=None, error_messages=Empty, execution_options=None, sanity_check=True, bind_group=None, **kwargs)[source]
Return type:

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

Parameters:

  • filters (StatementFilter | ColumnElement[bool])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • execution_options (dict[str, Any] | None)

  • sanity_check (bool)

  • bind_group (str | None)

  • kwargs (Any)

exists(*filters, load=None, error_messages=Empty, execution_options=None, bind_group=None, **kwargs)[source]
Return type:

bool

Parameters:

  • filters (StatementFilter | ColumnElement[bool])

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

  • kwargs (Any)

get(item_id, *, auto_expunge=None, statement=None, id_attribute=None, error_messages=Empty, load=None, execution_options=None, with_for_update=None, bind_group=None)[source]
Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Parameters:

  • item_id (Any)

  • auto_expunge (bool | None)

  • statement (Select | None)

  • id_attribute (str | InstrumentedAttribute[Any] | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • bind_group (str | None)

get_one(*filters, auto_expunge=None, statement=None, error_messages=Empty, load=None, execution_options=None, with_for_update=None, bind_group=None, **kwargs)[source]
Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Parameters:

  • filters (StatementFilter | ColumnElement[bool])

  • auto_expunge (bool | None)

  • statement (Select | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • bind_group (str | None)

  • kwargs (Any)

get_one_or_none(*filters, auto_expunge=None, statement=None, error_messages=Empty, load=None, execution_options=None, with_for_update=None, bind_group=None, **kwargs)[source]
Return type:

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

Parameters:

  • filters (StatementFilter | ColumnElement[bool])

  • auto_expunge (bool | None)

  • statement (Select | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • bind_group (str | None)

  • kwargs (Any)

get_or_upsert(*filters, match_fields=None, upsert=True, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=Empty, load=None, execution_options=None, bind_group=None, **kwargs)[source]
Return type:

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

Parameters:

  • filters (StatementFilter | ColumnElement[bool])

  • match_fields (list[str] | str | None)

  • upsert (bool)

  • attribute_names (Iterable[str] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • auto_refresh (bool | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

  • kwargs (Any)

get_and_update(*filters, match_fields=None, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=Empty, load=None, execution_options=None, bind_group=None, **kwargs)[source]
Return type:

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

Parameters:

  • filters (StatementFilter | ColumnElement[bool])

  • match_fields (list[str] | str | None)

  • attribute_names (Iterable[str] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • auto_refresh (bool | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

  • kwargs (Any)

count(*filters, statement=None, load=None, error_messages=Empty, execution_options=None, bind_group=None, **kwargs)[source]
Return type:

int

Parameters:

  • filters (StatementFilter | ColumnElement[bool])

  • statement (Select | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

  • kwargs (Any)

update(data, *, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, id_attribute=None, error_messages=Empty, load=None, execution_options=None, bind_group=None)[source]
Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Parameters:

  • data (ModelT)

  • attribute_names (Iterable[str] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • auto_refresh (bool | None)

  • id_attribute (str | InstrumentedAttribute[Any] | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

update_many(data, *, auto_commit=None, auto_expunge=None, error_messages=Empty, load=None, execution_options=None, bind_group=None)[source]
Return type:

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

Parameters:

  • data (list[ModelT])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

upsert(data, *, attribute_names=None, with_for_update=None, auto_expunge=None, auto_commit=None, auto_refresh=None, match_fields=None, error_messages=Empty, load=None, execution_options=None, bind_group=None)[source]
Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Parameters:

  • data (ModelT)

  • attribute_names (Iterable[str] | None)

  • with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])

  • auto_expunge (bool | None)

  • auto_commit (bool | None)

  • auto_refresh (bool | None)

  • match_fields (list[str] | str | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

upsert_many(data, *, auto_expunge=None, auto_commit=None, no_merge=False, match_fields=None, error_messages=Empty, load=None, execution_options=None, bind_group=None)[source]
Return type:

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

Parameters:

  • data (list[ModelT])

  • auto_expunge (bool | None)

  • auto_commit (bool | None)

  • no_merge (bool)

  • match_fields (list[str] | str | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

list_and_count(*filters, auto_expunge=None, statement=None, count_with_window_function=None, error_messages=Empty, load=None, execution_options=None, order_by=None, use_cache=True, bind_group=None, **kwargs)[source]
Return type:

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

Parameters:

  • filters (StatementFilter | ColumnElement[bool])

  • auto_expunge (bool | None)

  • statement (Select | None)

  • count_with_window_function (bool | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • order_by (list[tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any]] | tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any] | None)

  • use_cache (bool)

  • bind_group (str | None)

  • kwargs (Any)

list(*filters, auto_expunge=None, statement=None, error_messages=Empty, load=None, execution_options=None, order_by=None, use_cache=True, bind_group=None, **kwargs)[source]
Return type:

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

Parameters:

  • filters (StatementFilter | ColumnElement[bool])

  • auto_expunge (bool | None)

  • statement (Select | None)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • order_by (list[tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any]] | tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any] | None)

  • use_cache (bool)

  • bind_group (str | None)

  • kwargs (Any)

classmethod check_health(session)[source]
Return type:

bool

Parameters:

session (Session | scoped_session[Session])

class advanced_alchemy.repository.SQLAlchemySyncSlugRepository[source]

Bases: SQLAlchemySyncRepository[ModelT], SQLAlchemySyncSlugRepositoryProtocol[ModelT]

Extends the repository to include slug model features..

get_by_slug(slug, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None, **kwargs)[source]

Select record by slug value.

Return type:

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

Returns:

The model instance or None if not found.

Parameters:

  • slug (str)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • uniquify (bool | None)

  • bind_group (str | None)

  • kwargs (Any)

get_available_slug(value_to_slugify, **kwargs)[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 (Any) – stuff

  • value_to_slugify (str)

  • kwargs (Any)

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

Bases: SQLAlchemySyncRepositoryProtocol[ModelT], Protocol[ModelT]

Protocol for SQLAlchemy repositories that support slug-based operations.

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

Type Parameters:

ModelT: The SQLAlchemy model type this repository handles.

get_by_slug(slug, *, error_messages=Empty, load=None, execution_options=None, bind_group=None, **kwargs)[source]

Get a model instance by its slug.

Parameters:

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

  • error_messages (Union[ErrorMessages, type[Empty], None]) – Optional custom error message templates.

  • load (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Specification for eager loading of relationships.

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

  • bind_group (Optional[str]) – Optional routing group to use for the operation.

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

  • slug (str)

  • error_messages (ErrorMessages | type[Empty] | None)

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.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 | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

  • execution_options (dict[str, Any] | None)

  • bind_group (str | None)

  • kwargs (Any)

Returns:

The found model instance or None if not found.

Return type:

ModelT | None

get_available_slug(value_to_slugify, **kwargs)[source]

Generate a unique slug for a given value.

Parameters:

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

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

  • value_to_slugify (str)

  • kwargs (Any)

Returns:

A unique slug derived from the input value.

Return type:

str

advanced_alchemy.repository.get_instrumented_attr(model, key)[source]

Get an instrumented attribute from a model.

Parameters:
Returns:

The instrumented attribute from the model.

Return type:

sqlalchemy.orm.InstrumentedAttribute

advanced_alchemy.repository.model_from_dict(model, **kwargs)[source]

Create an ORM model instance from a dictionary of attributes.

This function recursively converts nested dictionaries into their corresponding SQLAlchemy model instances for relationship attributes.

Parameters:

  • model (type[TypeVar(ModelT, bound= base.ModelProtocol)]) – The SQLAlchemy model class to instantiate.

  • **kwargs (Any) – Keyword arguments containing model attribute values. For relationship attributes, values can be: - None: Sets the relationship to None - dict: Recursively converted to the related model instance - list[dict]: Each dict converted to related model instances - Model instance: Passed through unchanged

  • model (type[ModelT])

  • kwargs (Any)

Returns:

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

Return type:

ModelT

Example

Basic usage with nested relationships:

data = {
    "name": "John Doe",
    "profile": {"bio": "Developer"},
    "addresses": [
        {"street": "123 Main St"},
        {"street": "456 Oak Ave"},
    ],
}
user = model_from_dict(User, **data)
# user.profile is a Profile instance
# user.addresses is a list of Address instances