Single objects are transformed to the supplied schema type, and lists of objects are automatically transformed into an OffsetPagination response of the supplied schema type.

Parameters:: data¶ – A database model instance or row mapping. Type: ModelOrRowMappingT
Returns:: The converted schema object.

to_schema(data, total=None, filters=None, *, schema_type=None)[source]¶

Overloads:

self, data (ModelOrRowMappingT), schema_type (None) → ModelOrRowMappingT
self, data (Union[ModelProtocol, RowMapping, Row[Any], dict[str, Any]]), schema_type (type[ModelDTOT]) → ModelDTOT
self, data (ModelOrRowMappingT), total (Optional[int]), schema_type (None) → ModelOrRowMappingT
self, data (Union[ModelProtocol, RowMapping, Row[Any], dict[str, Any]]), total (Optional[int]), schema_type (type[ModelDTOT]) → ModelDTOT
self, data (ModelOrRowMappingT), total (Optional[int]), filters (Union[Sequence[Union[StatementFilter, ColumnElement[bool]]], Sequence[StatementFilter], None]), schema_type (None) → ModelOrRowMappingT
self, data (Union[ModelProtocol, RowMapping, Row[Any], dict[str, Any]]), total (Optional[int]), filters (Union[Sequence[Union[StatementFilter, ColumnElement[bool]]], Sequence[StatementFilter], None]), schema_type (type[ModelDTOT]) → ModelDTOT
self, data (Sequence[Row[Any]]), total (int) → OffsetPagination[Row[Any]]
self, data (Sequence[ModelOrRowMappingT]), schema_type (None) → OffsetPagination[ModelOrRowMappingT]
self, data (Union[Sequence[ModelProtocol], Sequence[RowMapping], Sequence[Row[Any]], Sequence[dict[str, Any]]]), schema_type (type[ModelDTOT]) → OffsetPagination[ModelDTOT]
self, data (Sequence[ModelOrRowMappingT]), total (Optional[int]), filters (Union[Sequence[Union[StatementFilter, ColumnElement[bool]]], Sequence[StatementFilter], None]), schema_type (None) → OffsetPagination[ModelOrRowMappingT]
self, data (Union[Sequence[ModelProtocol], Sequence[RowMapping], Sequence[Row[Any]], Sequence[dict[str, Any]]]), total (Optional[int]), filters (Union[Sequence[Union[StatementFilter, ColumnElement[bool]]], Sequence[StatementFilter], None]), schema_type (type[ModelDTOT]) → OffsetPagination[ModelDTOT]

Parameters:

data (Union[ModelOrRowMappingT, Sequence[ModelOrRowMappingT], ModelProtocol, Sequence[ModelProtocol], RowMapping, Sequence[RowMapping], Row[Any], Sequence[Row[Any]], dict[str, Any], Sequence[dict[str, Any]]])
total (Optional[int])
filters (Union[Sequence[Union[StatementFilter, ColumnElement[bool]]], Sequence[StatementFilter], None])
schema_type (Optional[type[ModelDTOT]])

Return type:

Union[ModelOrRowMappingT, OffsetPagination[ModelOrRowMappingT], ModelDTOT, OffsetPagination[ModelDTOT]]

Convert the object to a response schema.

When schema_type is None, the model is returned with no conversion.

Parameters:

data¶ – The return from one of the service calls. Type: ModelOrRowMappingT
total¶ – The total number of rows in the data.
filters¶ – ColumnElement Collection of route filters.
schema_type¶ – ModelDTOT Optional schema type to convert the data to
data (Union[ModelOrRowMappingT, Sequence[ModelOrRowMappingT], ModelProtocol, Sequence[ModelProtocol], RowMapping, Sequence[RowMapping], Row[Any], Sequence[Row[Any]], dict[str, Any], Sequence[dict[str, Any]]])
total (Optional[int])
filters (Union[Sequence[Union[StatementFilter, ColumnElement[bool]]], Sequence[StatementFilter], None])
schema_type (Optional[type[ModelDTOT]])

Raises:

AdvancedAlchemyError – If schema_type is not a valid Pydantic, Msgspec, or attrs schema and all libraries are not installed.

Returns:

Return type:

Union[ModelOrRowMappingT, OffsetPagination[ModelOrRowMappingT], ModelDTOT, OffsetPagination[ModelDTOT]]

class advanced_alchemy.service.SQLAlchemyAsyncQueryService[source]¶

Bases: ResultConverter

Simple service to execute the basic Query repository..

__init__(session, **repo_kwargs)[source]¶

Configure the service object.

Parameters:

session¶ (Union[AsyncSession, async_scoped_session[AsyncSession]]) – Session managing the unit-of-work for the operation.
**repo_kwargs¶ (Any) – Optional configuration values to pass into the repository
session (AsyncSession | async_scoped_session[AsyncSession])
repo_kwargs (Any)

Return type:

None

classmethod new(cls, session=None, config=None)[source]¶

Context manager that returns instance of service object.

Handles construction of the database session._create_select_for_model

Raises:

AdvancedAlchemyError – If no configuration or session is provided.

Yields:

The service object instance.

Parameters:

session (AsyncSession | async_scoped_session[AsyncSession] | None)
config (SQLAlchemyAsyncConfig | None)

Return type:

AsyncIterator[Self]

class advanced_alchemy.service.SQLAlchemyAsyncRepositoryReadService[source]¶

Bases: ResultConverter, Generic[ModelT, SQLAlchemyAsyncRepositoryT]

Service object that operates on a repository object.

repository_type: type[SQLAlchemyAsyncRepositoryT]¶: Type of the repository to use.

loader_options: ClassVar[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.

execution_options: ClassVar[dict[str, Any] | None] = None¶: Default execution options for the repository.

match_fields: ClassVar[List[str] | str | None] = None¶: List of dialects that prefer to use field.id = ANY(:1) instead of field.id IN (...).

uniquify: ClassVar[bool] = False¶: Optionally apply the unique() method to results before returning.

count_with_window_function: ClassVar[bool] = True¶: Use an analytical window function to count results. This allows the count to be performed in a single query.

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

Configure the service object.

Parameters:

session¶ (Union[AsyncSession, async_scoped_session[AsyncSession]]) – Session managing the unit-of-work for the operation.
statement¶ (Optional[Select]) – To facilitate customization of the underlying select query.
auto_expunge¶ (bool) – Remove object from session before returning.
auto_refresh¶ (bool) – Refresh object from session before returning.
auto_commit¶ (bool) – Commit objects before returning.
order_by¶ (Union[List[Union[tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any]]], tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any], None]) – Set default order options for queries.
error_messages¶ (Union[ErrorMessages, type[Empty], None]) – A set of custom error messages to use for operations
wrap_exceptions¶ (bool) – Wrap exceptions in a RepositoryError
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.
count_with_window_function¶ (Optional[bool]) – When false, list and count will use two queries instead of an analytical window function.
**repo_kwargs¶ (Any) – passed as keyword args to repo instantiation.
session (AsyncSession | async_scoped_session[AsyncSession])
statement (Select | None)
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)
error_messages (ErrorMessages | type[Empty] | None)
wrap_exceptions (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)
uniquify (bool | None)
count_with_window_function (bool | None)
repo_kwargs (Any)

Return type:

None

property repository: SQLAlchemyAsyncRepositoryT¶

Return the repository instance.

Raises:: ImproperConfigurationError – If the repository is not initialized.
Returns:: The repository instance.

property model_type: type[ModelT][source]¶: Return the model type.

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

Count of records returned by query.

Parameters:

*filters¶ (Union[StatementFilter, ColumnElement[bool]]) – arguments for filtering.
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.
bind_group¶ (Optional[str]) – The bind group to use for the operation.
**kwargs¶ (Any) – key value pairs of filter types.
filters (StatementFilter | ColumnElement[bool])
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)
bind_group (str | None)
kwargs (Any)

Return type:

int

Returns:

A count of the collection, filtered, but ignoring pagination.

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

Wrap repository exists operation.

Parameters:

*filters¶ (Union[StatementFilter, ColumnElement[bool]]) – Types for specific filtering operations.
error_messages¶ (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients
load¶ (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set 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]) – The bind group to use for the operation.
**kwargs¶ (Any) – Keyword arguments for attribute based filtering.
filters (StatementFilter | ColumnElement[bool])
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:

bool

Returns:

Representation of instance with identifier item_id.

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

Wrap repository scalar operation.

Parameters:

item_id¶ (Union[Any, tuple[Any, ...], dict[str, Any]]) – Identifier of instance to be retrieved. For single primary key models, pass a scalar value (int, str, UUID, etc.). For composite primary key models, pass a tuple of values in column order, or a dict mapping attribute names to values.
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. Only applicable for single primary key models.
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]) – The bind group to use for the operation.
item_id (Any | tuple[Any, ...] | dict[str, Any])
statement (Select | None)
id_attribute (str | InstrumentedAttribute[Any] | 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:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

Representation of instance with identifier item_id.

Examples

# Single primary key >>> user = await service.get(123)

# Composite primary key (tuple format) >>> assignment = await service.get((user_id, role_id))

# Composite primary key (dict format) >>> assignment = await service.get({“user_id”: 1, “role_id”: 5})

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

Wrap repository scalar operation.

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 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.
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]) – The bind group to use for the operation.
**kwargs¶ (Any) – Identifier of the instance to be retrieved.
filters (StatementFilter | ColumnElement[bool])
statement (Select | 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)
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:

Representation of instance with identifier item_id.

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

Wrap repository scalar operation.

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 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.
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]) – The bind group to use for the operation.
**kwargs¶ (Any) – Identifier of the instance to be retrieved.
filters (StatementFilter | ColumnElement[bool])
statement (Select | 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)
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:

Representation of instance with identifier item_id.

async to_model_on_create(data)[source]¶

Convenience method to allow for custom behavior on create.

Parameters:

data¶ – The data to be converted to a model.
data (ModelDictT[ModelT])

Returns:

The data to be converted to a model.

Return type:

ModelDictT[ModelT]

async to_model_on_update(data)[source]¶

Convenience method to allow for custom behavior on update.

Parameters:

data¶ – The data to be converted to a model.
data (ModelDictT[ModelT])

Returns:

The data to be converted to a model.

Return type:

ModelDictT[ModelT]

async to_model_on_delete(data)[source]¶

Convenience method to allow for custom behavior on delete.

Parameters:

data¶ – The data to be converted to a model.
data (ModelDictT[ModelT])

Returns:

The data to be converted to a model.

Return type:

ModelDictT[ModelT]

async to_model_on_upsert(data)[source]¶

Convenience method to allow for custom behavior on upsert.

Parameters:

data¶ – The data to be converted to a model.
data (ModelDictT[ModelT])

Returns:

The data to be converted to a model.

Return type:

ModelDictT[ModelT]

async to_model(data, operation=None)[source]¶

Parse and Convert input into a model.

Parameters:

data¶ – Representations to be created.
operation¶ – Optional operation flag so that you can provide behavior based on CRUD operation
data (ModelDictT[ModelT])
operation (str | None)

Returns:

Representation of created instances.

Return type:

ModelT

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 of records and total count returned by query.

Parameters:

*filters¶ (Union[StatementFilter, ColumnElement[bool]]) – Types for specific filtering operations.
statement¶ (Optional[Select]) – To facilitate customization of the underlying select query.
auto_expunge¶ (Optional[bool]) – Remove object from session before returning.
count_with_window_function¶ (Optional[bool]) – When false, list and count will use two queries instead of an analytical window function.
order_by¶ (Union[List[Union[tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any]]], tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any], None]) – Set default order options for queries.
error_messages¶ (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients
load¶ (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded
execution_options¶ (Optional[dict[str, Any]]) – Set default execution options
uniquify¶ (Optional[bool]) – Optionally apply the unique() method to results before returning.
use_cache¶ (bool) – Whether to use the repository cache for this query.
bind_group¶ (Optional[str]) – Optional routing group to use for the operation.
**kwargs¶ (Any) – Instance attribute value filters.
filters (StatementFilter | ColumnElement[bool])
statement (Select | None)
auto_expunge (bool | None)
count_with_window_function (bool | 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)
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)
use_cache (bool)
bind_group (str | None)
kwargs (Any)

Return type:

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

Returns:

List of instances and count of total collection, ignoring pagination.

classmethod new(cls, session=None, statement=None, config=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None, uniquify=None, count_with_window_function=None)[source]¶

Context manager that returns instance of service object.

Handles construction of the database session._create_select_for_model

Raises:

AdvancedAlchemyError – If no configuration or session is provided.

Yields:

The service object instance.

Parameters:

session (AsyncSession | async_scoped_session[AsyncSession] | None)
statement (Select | None)
config (SQLAlchemyAsyncConfig | 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)
count_with_window_function (bool | None)

Return type:

AsyncIterator[Self]

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

Wrap repository scalars operation.

Parameters:

*filters¶ (Union[StatementFilter, ColumnElement[bool]]) – Types for specific filtering operations.
auto_expunge¶ (Optional[bool]) – Remove object from session before returning.
statement¶ (Optional[Select]) – To facilitate customization of the underlying select query.
order_by¶ (Union[List[Union[tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any]]], tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any], None]) – Set default order options for queries.
error_messages¶ (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients
load¶ (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set 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.
use_cache¶ (bool) – Whether to use the repository cache for this query.
bind_group¶ (Optional[str]) – Optional routing group to use for the operation.
**kwargs¶ (Any) – Instance attribute value filters.
filters (StatementFilter | ColumnElement[bool])
statement (Select | None)
auto_expunge (bool | 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)
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)
use_cache (bool)
bind_group (str | None)
kwargs (Any)

Return type:

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

Returns:

The list of instances retrieved from the repository.

class advanced_alchemy.service.SQLAlchemyAsyncRepositoryService[source]¶

Bases: SQLAlchemyAsyncRepositoryReadService[ModelT, SQLAlchemyAsyncRepositoryT], Generic[ModelT, SQLAlchemyAsyncRepositoryT]

Service object that operates on a repository object.

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

Wrap repository instance creation.

Parameters:

data¶ – Representation to be created.
auto_expunge¶ – Remove object from session before returning.
auto_refresh¶ – Refresh object from session before returning.
auto_commit¶ – Commit objects before returning.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
bind_group¶ – Optional routing group to use for the operation.
data (ModelDictT[ModelT])
auto_commit (bool | None)
auto_expunge (bool | None)
auto_refresh (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
bind_group (str | None)

Returns:

Representation of created instance.

Return type:

ModelT

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

Wrap repository bulk instance creation.

Parameters:

data¶ – Representations to be created.
auto_expunge¶ – Remove object from session before returning.
auto_commit¶ – Commit objects before returning.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
bind_group¶ – Optional routing group to use for the operation.
data (BulkModelDictT[ModelT])
auto_commit (bool | None)
auto_expunge (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
bind_group (str | None)

Returns:

Representation of created instances.

Return type:

Sequence[ModelT]

async update(data, item_id=None, *, 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]¶

Wrap repository update operation.

Parameters:

data¶ – Representation to be updated.
item_id¶ – Identifier of item to be updated.
attribute_names¶ – an iterable of attribute names to pass into the update method.
with_for_update¶ – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT
auto_expunge¶ – Remove object from session before returning.
auto_refresh¶ – Refresh object from session before returning.
auto_commit¶ – Commit objects before returning.
id_attribute¶ – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
uniquify¶ – Optionally apply the unique() method to results before returning.
bind_group¶ – Optional routing group to use for the operation.
data (ModelDictT[ModelT])
item_id (Any | None)
attribute_names (Iterable[str] | None)
with_for_update (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)

Raises:

RepositoryError – If no configuration or session is provided.

Returns:

Updated representation.

Return type:

ModelT

async update_many(data, *, auto_commit=None, auto_expunge=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None)[source]¶

Wrap repository bulk instance update.

Parameters:

data¶ – Representations to be updated.
auto_expunge¶ – Remove object from session before returning.
auto_commit¶ – Commit objects before returning.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
uniquify¶ – Optionally apply the unique() method to results before returning.
bind_group¶ – Optional routing group to use for the operation.
data (BulkModelDictT[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)

Returns:

Representation of updated instances.

Return type:

Sequence[ModelT]

async upsert(data, item_id=None, *, 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]¶

Wrap repository upsert operation.

Parameters:

data¶ – Instance to update existing, or be created. Identifier used to determine if an existing instance exists is the value of an attribute on data named as value of self.id_attribute.
item_id¶ – Identifier of the object for upsert.
attribute_names¶ – an iterable of attribute names to pass into the update method.
with_for_update¶ – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT
auto_expunge¶ – Remove object from session before returning.
auto_refresh¶ – Refresh object from session before returning.
auto_commit¶ – Commit objects before returning.
match_fields¶ – a list of keys to use to match the existing model. When empty, all fields are matched.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
uniquify¶ – Optionally apply the unique() method to results before returning.
bind_group¶ – Optional routing group to use for the operation.
data (ModelDictT[ModelT])
item_id (Any | None)
attribute_names (Iterable[str] | None)
with_for_update (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)

Returns:

Updated or created representation.

Return type:

ModelT

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

Wrap repository upsert operation.

Parameters:

data¶ – Instance to update existing, or be created.
auto_expunge¶ – Remove object from session before returning.
auto_commit¶ – Commit objects before returning.
no_merge¶ – Skip the usage of optimized Merge statements (reserved for future use)
match_fields¶ – a list of keys to use to match the existing model. When empty, all fields are matched.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
uniquify¶ – Optionally apply the unique() method to results before returning.
bind_group¶ – Optional routing group to use for the operation.
data (BulkModelDictT[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)

Returns:

Updated or created representation.

Return type:

Sequence[ModelT]

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

Wrap repository instance creation.

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, perform an update operation on the model.
create¶ – Should a model be created. If no model is found, an exception is raised.
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 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.
**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:

Representation of created instance.

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

Wrap repository instance creation.

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

Representation of updated instance.

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

Wrap repository delete operation.

Parameters:

item_id¶ (Union[Any, tuple[Any, ...], dict[str, Any]]) – Identifier of instance to be deleted. For single primary key models, pass a scalar value (int, str, UUID, etc.). For composite primary key models, pass a tuple of values in column order, or a dict mapping attribute names to values.
auto_commit¶ (Optional[bool]) – Commit objects before returning.
auto_expunge¶ (Optional[bool]) – Remove object from session 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. Only applicable for single primary key models.
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.
item_id (Any | tuple[Any, ...] | dict[str, 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:

Representation of the deleted instance.

Examples

# Single primary key >>> deleted_user = await service.delete(123)

# Composite primary key (tuple format) >>> deleted = await service.delete((user_id, role_id))

# Composite primary key (dict format) >>> deleted = await service.delete({“user_id”: 1, “role_id”: 5})

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

Wrap repository bulk instance deletion.

Parameters:

item_ids¶ (List[Union[Any, tuple[Any, ...], dict[str, Any]]]) – List of identifiers of instances to be deleted. For single primary key models, pass a list of scalar values. For composite primary key models, pass a list of tuples or dicts.
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. Only applicable for single primary key models.
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 to use for the operation.
item_ids (List[Any | tuple[Any, ...] | dict[str, 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:

Representation of removed instances.

Examples

# Single primary key >>> deleted = await service.delete_many([1, 2, 3])

# Composite primary key (tuple format) >>> deleted = await service.delete_many( … [ … (user_id_1, role_id_1), … (user_id_2, role_id_2), … ] … )

# Composite primary key (dict format) >>> deleted = await service.delete_many( … [ … {“user_id”: 1, “role_id”: 5}, … {“user_id”: 1, “role_id”: 6}, … ] … )

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

Wrap repository scalars operation.

Parameters:

*filters¶ (Union[StatementFilter, ColumnElement[bool]]) – Types for specific filtering operations.
auto_expunge¶ (Optional[bool]) – Remove object from session before returning.
auto_commit¶ (Optional[bool]) – Commit objects before returning.
error_messages¶ (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients
sanity_check¶ (bool) – When true, the length of selected instances is compared to the deleted row count
load¶ (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded
execution_options¶ (Optional[dict[str, Any]]) – Set default execution options
uniquify¶ (Optional[bool]) – Optionally apply the unique() method to results before returning.
bind_group¶ (Optional[str]) – Optional routing group to use for the operation.
**kwargs¶ (Any) – Instance attribute value filters.
filters (StatementFilter | ColumnElement[bool])
auto_commit (bool | None)
auto_expunge (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
sanity_check (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)
uniquify (bool | None)
bind_group (str | None)
kwargs (Any)

Return type:

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

Returns:

The list of instances deleted from the repository.

class advanced_alchemy.service.SQLAlchemySyncQueryService[source]¶

Bases: ResultConverter

Simple service to execute the basic Query repository..

__init__(session, **repo_kwargs)[source]¶

Configure the service object.

Parameters:

session¶ (Union[Session, scoped_session[Session]]) – Session managing the unit-of-work for the operation.
**repo_kwargs¶ (Any) – Optional configuration values to pass into the repository
session (Session | scoped_session[Session])
repo_kwargs (Any)

Return type:

None

classmethod new(cls, session=None, config=None)[source]¶

Context manager that returns instance of service object.

Handles construction of the database session._create_select_for_model

Raises:

AdvancedAlchemyError – If no configuration or session is provided.

Yields:

The service object instance.

Parameters:

session (Session | scoped_session[Session] | None)
config (SQLAlchemySyncConfig | None)

Return type:

Iterator[Self]

class advanced_alchemy.service.SQLAlchemySyncRepositoryReadService[source]¶

Bases: ResultConverter, Generic[ModelT, SQLAlchemySyncRepositoryT]

Service object that operates on a repository object.

repository_type: type[SQLAlchemySyncRepositoryT]¶: Type of the repository to use.

loader_options: ClassVar[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.

execution_options: ClassVar[dict[str, Any] | None] = None¶: Default execution options for the repository.

match_fields: ClassVar[List[str] | str | None] = None¶: List of dialects that prefer to use field.id = ANY(:1) instead of field.id IN (...).

uniquify: ClassVar[bool] = False¶: Optionally apply the unique() method to results before returning.

count_with_window_function: ClassVar[bool] = True¶: Use an analytical window function to count results. This allows the count to be performed in a single query.

Configure the service object.

Parameters:

session¶ (Union[Session, scoped_session[Session]]) – Session managing the unit-of-work for the operation.
statement¶ (Optional[Select]) – To facilitate customization of the underlying select query.
auto_expunge¶ (bool) – Remove object from session before returning.
auto_refresh¶ (bool) – Refresh object from session before returning.
auto_commit¶ (bool) – Commit objects before returning.
order_by¶ (Union[List[Union[tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any]]], tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any], None]) – Set default order options for queries.
error_messages¶ (Union[ErrorMessages, type[Empty], None]) – A set of custom error messages to use for operations
wrap_exceptions¶ (bool) – Wrap exceptions in a RepositoryError
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.
count_with_window_function¶ (Optional[bool]) – When false, list and count will use two queries instead of an analytical window function.
**repo_kwargs¶ (Any) – passed as keyword args to repo instantiation.
session (Session | scoped_session[Session])
statement (Select | None)
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)
error_messages (ErrorMessages | type[Empty] | None)
wrap_exceptions (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)
uniquify (bool | None)
count_with_window_function (bool | None)
repo_kwargs (Any)

Return type:

None

property repository: SQLAlchemySyncRepositoryT¶

Return the repository instance.

Raises:: ImproperConfigurationError – If the repository is not initialized.
Returns:: The repository instance.

property model_type: type[ModelT][source]¶: Return the model type.

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

Count of records returned by query.

Parameters:

*filters¶ (Union[StatementFilter, ColumnElement[bool]]) – arguments for filtering.
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.
bind_group¶ (Optional[str]) – The bind group to use for the operation.
**kwargs¶ (Any) – key value pairs of filter types.
filters (StatementFilter | ColumnElement[bool])
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)
bind_group (str | None)
kwargs (Any)

Return type:

int

Returns:

A count of the collection, filtered, but ignoring pagination.

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

Wrap repository exists operation.

Parameters:

*filters¶ (Union[StatementFilter, ColumnElement[bool]]) – Types for specific filtering operations.
error_messages¶ (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients
load¶ (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set 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]) – The bind group to use for the operation.
**kwargs¶ (Any) – Keyword arguments for attribute based filtering.
filters (StatementFilter | ColumnElement[bool])
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:

bool

Returns:

Representation of instance with identifier item_id.

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

Wrap repository scalar operation.

Parameters:

item_id¶ (Union[Any, tuple[Any, ...], dict[str, Any]]) – Identifier of instance to be retrieved. For single primary key models, pass a scalar value (int, str, UUID, etc.). For composite primary key models, pass a tuple of values in column order, or a dict mapping attribute names to values.
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. Only applicable for single primary key models.
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]) – The bind group to use for the operation.
item_id (Any | tuple[Any, ...] | dict[str, Any])
statement (Select | None)
id_attribute (str | InstrumentedAttribute[Any] | 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:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

Representation of instance with identifier item_id.

Examples

# Single primary key >>> user = await service.get(123)

# Composite primary key (tuple format) >>> assignment = await service.get((user_id, role_id))

# Composite primary key (dict format) >>> assignment = await service.get({“user_id”: 1, “role_id”: 5})

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

Wrap repository scalar operation.

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 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.
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]) – The bind group to use for the operation.
**kwargs¶ (Any) – Identifier of the instance to be retrieved.
filters (StatementFilter | ColumnElement[bool])
statement (Select | 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)
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:

Representation of instance with identifier item_id.

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

Wrap repository scalar operation.

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 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.
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]) – The bind group to use for the operation.
**kwargs¶ (Any) – Identifier of the instance to be retrieved.
filters (StatementFilter | ColumnElement[bool])
statement (Select | 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)
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:

Representation of instance with identifier item_id.

to_model_on_create(data)[source]¶

Convenience method to allow for custom behavior on create.

Parameters:

data¶ – The data to be converted to a model.
data (ModelDictT[ModelT])

Returns:

The data to be converted to a model.

Return type:

ModelDictT[ModelT]

to_model_on_update(data)[source]¶

Convenience method to allow for custom behavior on update.

Parameters:

data¶ – The data to be converted to a model.
data (ModelDictT[ModelT])

Returns:

The data to be converted to a model.

Return type:

ModelDictT[ModelT]

to_model_on_delete(data)[source]¶

Convenience method to allow for custom behavior on delete.

Parameters:

data¶ – The data to be converted to a model.
data (ModelDictT[ModelT])

Returns:

The data to be converted to a model.

Return type:

ModelDictT[ModelT]

to_model_on_upsert(data)[source]¶

Convenience method to allow for custom behavior on upsert.

Parameters:

data¶ – The data to be converted to a model.
data (ModelDictT[ModelT])

Returns:

The data to be converted to a model.

Return type:

ModelDictT[ModelT]

to_model(data, operation=None)[source]¶

Parse and Convert input into a model.

Parameters:

data¶ – Representations to be created.
operation¶ – Optional operation flag so that you can provide behavior based on CRUD operation
data (ModelDictT[ModelT])
operation (str | None)

Returns:

Representation of created instances.

Return type:

ModelT

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 of records and total count returned by query.

Parameters:

*filters¶ (Union[StatementFilter, ColumnElement[bool]]) – Types for specific filtering operations.
statement¶ (Optional[Select]) – To facilitate customization of the underlying select query.
auto_expunge¶ (Optional[bool]) – Remove object from session before returning.
count_with_window_function¶ (Optional[bool]) – When false, list and count will use two queries instead of an analytical window function.
order_by¶ (Union[List[Union[tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any]]], tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any], None]) – Set default order options for queries.
error_messages¶ (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients
load¶ (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set relationships to be loaded
execution_options¶ (Optional[dict[str, Any]]) – Set default execution options
uniquify¶ (Optional[bool]) – Optionally apply the unique() method to results before returning.
use_cache¶ (bool) – Whether to use the repository cache for this query.
bind_group¶ (Optional[str]) – Optional routing group to use for the operation.
**kwargs¶ (Any) – Instance attribute value filters.
filters (StatementFilter | ColumnElement[bool])
statement (Select | None)
auto_expunge (bool | None)
count_with_window_function (bool | 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)
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)
use_cache (bool)
bind_group (str | None)
kwargs (Any)

Return type:

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

Returns:

List of instances and count of total collection, ignoring pagination.

Context manager that returns instance of service object.

Handles construction of the database session._create_select_for_model

Raises:

AdvancedAlchemyError – If no configuration or session is provided.

Yields:

The service object instance.

Parameters:

session (Session | scoped_session[Session] | None)
statement (Select | None)
config (SQLAlchemySyncConfig | 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)
count_with_window_function (bool | None)

Return type:

Iterator[Self]

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

Wrap repository scalars operation.

Parameters:

*filters¶ (Union[StatementFilter, ColumnElement[bool]]) – Types for specific filtering operations.
auto_expunge¶ (Optional[bool]) – Remove object from session before returning.
statement¶ (Optional[Select]) – To facilitate customization of the underlying select query.
order_by¶ (Union[List[Union[tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any]]], tuple[Union[str, InstrumentedAttribute[Any]], bool], UnaryExpression[Any], None]) – Set default order options for queries.
error_messages¶ (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients
load¶ (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set 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.
use_cache¶ (bool) – Whether to use the repository cache for this query.
bind_group¶ (Optional[str]) – Optional routing group to use for the operation.
**kwargs¶ (Any) – Instance attribute value filters.
filters (StatementFilter | ColumnElement[bool])
statement (Select | None)
auto_expunge (bool | 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)
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)
use_cache (bool)
bind_group (str | None)
kwargs (Any)

Return type:

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

Returns:

The list of instances retrieved from the repository.

class advanced_alchemy.service.SQLAlchemySyncRepositoryService[source]¶

Bases: SQLAlchemySyncRepositoryReadService[ModelT, SQLAlchemySyncRepositoryT], Generic[ModelT, SQLAlchemySyncRepositoryT]

Service object that operates on a repository object.

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

Wrap repository instance creation.

Parameters:

data¶ – Representation to be created.
auto_expunge¶ – Remove object from session before returning.
auto_refresh¶ – Refresh object from session before returning.
auto_commit¶ – Commit objects before returning.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
bind_group¶ – Optional routing group to use for the operation.
data (ModelDictT[ModelT])
auto_commit (bool | None)
auto_expunge (bool | None)
auto_refresh (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
bind_group (str | None)

Returns:

Representation of created instance.

Return type:

ModelT

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

Wrap repository bulk instance creation.

Parameters:

data¶ – Representations to be created.
auto_expunge¶ – Remove object from session before returning.
auto_commit¶ – Commit objects before returning.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
bind_group¶ – Optional routing group to use for the operation.
data (BulkModelDictT[ModelT])
auto_commit (bool | None)
auto_expunge (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
bind_group (str | None)

Returns:

Representation of created instances.

Return type:

Sequence[ModelT]

update(data, item_id=None, *, 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]¶

Wrap repository update operation.

Parameters:

data¶ – Representation to be updated.
item_id¶ – Identifier of item to be updated.
attribute_names¶ – an iterable of attribute names to pass into the update method.
with_for_update¶ – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT
auto_expunge¶ – Remove object from session before returning.
auto_refresh¶ – Refresh object from session before returning.
auto_commit¶ – Commit objects before returning.
id_attribute¶ – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
uniquify¶ – Optionally apply the unique() method to results before returning.
bind_group¶ – Optional routing group to use for the operation.
data (ModelDictT[ModelT])
item_id (Any | None)
attribute_names (Iterable[str] | None)
with_for_update (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)

Raises:

RepositoryError – If no configuration or session is provided.

Returns:

Updated representation.

Return type:

ModelT

update_many(data, *, auto_commit=None, auto_expunge=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, bind_group=None)[source]¶

Wrap repository bulk instance update.

Parameters:

data¶ – Representations to be updated.
auto_expunge¶ – Remove object from session before returning.
auto_commit¶ – Commit objects before returning.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
uniquify¶ – Optionally apply the unique() method to results before returning.
bind_group¶ – Optional routing group to use for the operation.
data (BulkModelDictT[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)

Returns:

Representation of updated instances.

Return type:

Sequence[ModelT]

upsert(data, item_id=None, *, 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]¶

Wrap repository upsert operation.

Parameters:

data¶ – Instance to update existing, or be created. Identifier used to determine if an existing instance exists is the value of an attribute on data named as value of self.id_attribute.
item_id¶ – Identifier of the object for upsert.
attribute_names¶ – an iterable of attribute names to pass into the update method.
with_for_update¶ – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT
auto_expunge¶ – Remove object from session before returning.
auto_refresh¶ – Refresh object from session before returning.
auto_commit¶ – Commit objects before returning.
match_fields¶ – a list of keys to use to match the existing model. When empty, all fields are matched.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
uniquify¶ – Optionally apply the unique() method to results before returning.
bind_group¶ – Optional routing group to use for the operation.
data (ModelDictT[ModelT])
item_id (Any | None)
attribute_names (Iterable[str] | None)
with_for_update (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)

Returns:

Updated or created representation.

Return type:

ModelT

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

Wrap repository upsert operation.

Parameters:

data¶ – Instance to update existing, or be created.
auto_expunge¶ – Remove object from session before returning.
auto_commit¶ – Commit objects before returning.
no_merge¶ – Skip the usage of optimized Merge statements (reserved for future use)
match_fields¶ – a list of keys to use to match the existing model. When empty, all fields are matched.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
uniquify¶ – Optionally apply the unique() method to results before returning.
bind_group¶ – Optional routing group to use for the operation.
data (BulkModelDictT[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)

Returns:

Updated or created representation.

Return type:

Sequence[ModelT]

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

Wrap repository instance creation.

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, perform an update operation on the model.
create¶ – Should a model be created. If no model is found, an exception is raised.
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 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.
**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:

Representation of created instance.

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

Wrap repository instance creation.

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

Representation of updated instance.

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

Wrap repository delete operation.

Parameters:

item_id¶ (Union[Any, tuple[Any, ...], dict[str, Any]]) – Identifier of instance to be deleted. For single primary key models, pass a scalar value (int, str, UUID, etc.). For composite primary key models, pass a tuple of values in column order, or a dict mapping attribute names to values.
auto_commit¶ (Optional[bool]) – Commit objects before returning.
auto_expunge¶ (Optional[bool]) – Remove object from session 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. Only applicable for single primary key models.
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.
item_id (Any | tuple[Any, ...] | dict[str, 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:

Representation of the deleted instance.

Examples

# Single primary key >>> deleted_user = await service.delete(123)

# Composite primary key (tuple format) >>> deleted = await service.delete((user_id, role_id))

# Composite primary key (dict format) >>> deleted = await service.delete({“user_id”: 1, “role_id”: 5})

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

Wrap repository bulk instance deletion.

Parameters:

item_ids¶ (List[Union[Any, tuple[Any, ...], dict[str, Any]]]) – List of identifiers of instances to be deleted. For single primary key models, pass a list of scalar values. For composite primary key models, pass a list of tuples or dicts.
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. Only applicable for single primary key models.
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 to use for the operation.
item_ids (List[Any | tuple[Any, ...] | dict[str, 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:

Representation of removed instances.

Examples

# Single primary key >>> deleted = await service.delete_many([1, 2, 3])

# Composite primary key (tuple format) >>> deleted = await service.delete_many( … [ … (user_id_1, role_id_1), … (user_id_2, role_id_2), … ] … )

# Composite primary key (dict format) >>> deleted = await service.delete_many( … [ … {“user_id”: 1, “role_id”: 5}, … {“user_id”: 1, “role_id”: 6}, … ] … )

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

Wrap repository scalars operation.

Parameters:

*filters¶ (Union[StatementFilter, ColumnElement[bool]]) – Types for specific filtering operations.
auto_expunge¶ (Optional[bool]) – Remove object from session before returning.
auto_commit¶ (Optional[bool]) – Commit objects before returning.
error_messages¶ (Union[ErrorMessages, type[Empty], None]) – An optional dictionary of templates to use for friendlier error messages to clients
sanity_check¶ (bool) – When true, the length of selected instances is compared to the deleted row count
load¶ (Union[Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], Sequence[Union[_AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any]]]]], _AbstractLoad, Literal['*'], InstrumentedAttribute[Any], RelationshipProperty[Any], MapperProperty[Any], ExecutableOption, Sequence[ExecutableOption], None]) – Set default relationships to be loaded
execution_options¶ (Optional[dict[str, Any]]) – Set default execution options
uniquify¶ (Optional[bool]) – Optionally apply the unique() method to results before returning.
bind_group¶ (Optional[str]) – Optional routing group to use for the operation.
**kwargs¶ (Any) – Instance attribute value filters.
filters (StatementFilter | ColumnElement[bool])
auto_commit (bool | None)
auto_expunge (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
sanity_check (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)
uniquify (bool | None)
bind_group (str | None)
kwargs (Any)

Return type:

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

Returns:

The list of instances deleted from the repository.

advanced_alchemy.service.fields(cls)[source]¶

Return the tuple of attrs attributes for a class.

The tuple also allows accessing the fields by their names (see below for examples).

Parameters:

cls¶ (type[AttrsInstance]) – Class to introspect.

Raises:

TypeError – If cls is not a class.
attrs.exceptions.NotAnAttrsClassError – If cls is not an attrs class.

Return type:

typing.Any

Returns:

tuple (with name accessors) of attrs.Attribute

Changed in version 16.2.0: Returned tuple allows accessing the fields by name.

Changed in version 23.1.0: Add support for generic classes.

advanced_alchemy.service.find_filter(filter_type, filters)[source]¶

Get the filter specified by filter type from the filters.

Parameters:

filter_type¶ – The type of filter to find.
filters¶ – filter types to apply to the query
filter_type (type[FilterTypeT])
filters (Union[Sequence[Union[StatementFilter, ColumnElement[bool]]], Sequence[StatementFilter]])

Returns:

The match filter instance or None

Return type:

Union[FilterTypeT, None]

advanced_alchemy.service.is_attrs_instance(obj)[source]¶

Check if a value is an attrs class instance.

Parameters:

obj¶ (Any) – Value to check.
obj (Any)

Return type:

TypeGuard[AttrsLike]

Returns:

bool

advanced_alchemy.service.is_attrs_instance_with_field(v, field_name)[source]¶

Check if an attrs instance has a specific field.

Parameters:

v¶ (Any) – Value to check.
field_name¶ (str) – Field name to check for.
v (Any)
field_name (str)

Return type:

TypeGuard[AttrsLike]

Returns:

bool

advanced_alchemy.service.is_attrs_instance_without_field(v, field_name)[source]¶

Check if an attrs instance does not have a specific field.

Parameters:

v¶ (Any) – Value to check.
field_name¶ (str) – Field name to check for.
v (Any)
field_name (str)

Return type:

TypeGuard[AttrsLike]

Returns:

bool

advanced_alchemy.service.is_attrs_schema(cls)[source]¶

Check if a class type is an attrs schema.

Parameters:

cls¶ (Any) – Class to check.
cls (Any)

Return type:

TypeGuard[type[AttrsLike]]

Returns:

bool

advanced_alchemy.service.is_dict(v)[source]¶

Check if a value is a dictionary.

Parameters:

v¶ (Any) – Value to check.
v (Any)

Return type:

TypeGuard[dict[str, Any]]

Returns:

bool

advanced_alchemy.service.is_dict_with_field(v, field_name)[source]¶

Check if a dictionary has a specific field.

Parameters:

v¶ (Any) – Value to check.
field_name¶ (str) – Field name to check for.
v (Any)
field_name (str)

Return type:

TypeGuard[dict[str, Any]]

Returns:

bool

advanced_alchemy.service.is_dict_without_field(v, field_name)[source]¶

Check if a dictionary does not have a specific field.

Parameters:

v¶ (Any) – Value to check.
field_name¶ (str) – Field name to check for.
v (Any)
field_name (str)

Return type:

TypeGuard[dict[str, Any]]

Returns:

bool

advanced_alchemy.service.is_dto_data(v)[source]¶

Check if a value is a Litestar DTOData object.

Parameters:

v¶ (Any) – Value to check.
v (Any)

Return type:

TypeGuard[DTODataLike[Any]]

Returns:

bool

advanced_alchemy.service.is_msgspec_struct(v)[source]¶

Check if a value is a msgspec struct.

Parameters:

v¶ (Any) – Value to check.
v (Any)

Return type:

TypeGuard[StructLike]

Returns:

bool

advanced_alchemy.service.is_msgspec_struct_with_field(v, field_name)[source]¶

Check if a msgspec struct has a specific field.

Parameters:

v¶ (Any) – Value to check.
field_name¶ (str) – Field name to check for.
v (Any)
field_name (str)

Return type:

TypeGuard[StructLike]

Returns:

bool

advanced_alchemy.service.is_msgspec_struct_without_field(v, field_name)[source]¶

Check if a msgspec struct does not have a specific field.

Parameters:

v¶ (Any) – Value to check.
field_name¶ (str) – Field name to check for.
v (Any)
field_name (str)

Return type:

TypeGuard[StructLike]

Returns:

bool

advanced_alchemy.service.is_pydantic_model(v)[source]¶

Check if a value is a pydantic model.

Parameters:

v¶ (Any) – Value to check.
v (Any)

Return type:

TypeGuard[BaseModelLike]

Returns:

bool

advanced_alchemy.service.is_pydantic_model_with_field(v, field_name)[source]¶

Check if a pydantic model has a specific field.

Parameters:

v¶ (Any) – Value to check.
field_name¶ (str) – Field name to check for.
v (Any)
field_name (str)

Return type:

TypeGuard[BaseModelLike]

Returns:

bool

advanced_alchemy.service.is_pydantic_model_without_field(v, field_name)[source]¶

Check if a pydantic model does not have a specific field.

Parameters:

v¶ (Any) – Value to check.
field_name¶ (str) – Field name to check for.
v (Any)
field_name (str)

Return type:

TypeGuard[BaseModelLike]

Returns:

bool

advanced_alchemy.service.is_schema(v)[source]¶

Check if a value is a msgspec Struct, Pydantic model, or attrs instance.

SQLModel table=True models are excluded — they are ORM-mapped models, not schemas that should be decomposed to dicts.

Parameters:

v¶ (Any) – Value to check.
v (Any)

Return type:

TypeGuard[Union[StructLike, BaseModelLike, AttrsLike]]

Returns:

bool

advanced_alchemy.service.is_schema_or_dict(v)[source]¶

Check if a value is a msgspec Struct, Pydantic model, attrs class, or dict.

Parameters:

v¶ – Value to check.
v (Any)

Returns:

bool

Return type:

TypeGuard[Union[SupportedSchemaModel, dict[str, Any]]]

advanced_alchemy.service.is_schema_or_dict_with_field(v, field_name)[source]¶

Check if a value is a msgspec Struct, Pydantic model, attrs instance, or dict with a specific field.

Parameters:

v¶ – Value to check.
field_name¶ – Field name to check for.
v (Any)
field_name (str)

Returns:

bool

Return type:

TypeGuard[Union[SupportedSchemaModel, dict[str, Any]]]

advanced_alchemy.service.is_schema_or_dict_without_field(v, field_name)[source]¶

Check if a value is a msgspec Struct, Pydantic model, attrs instance, or dict without a specific field.

Parameters:

v¶ – Value to check.
field_name¶ – Field name to check for.
v (Any)
field_name (str)

Returns:

bool

Return type:

TypeGuard[Union[SupportedSchemaModel, dict[str, Any]]]

advanced_alchemy.service.is_schema_with_field(v, field_name)[source]¶

Check if a value is a msgspec Struct, Pydantic model, or attrs instance with a specific field.

SQLModel table=True models are excluded.

Parameters:

v¶ (Any) – Value to check.
field_name¶ (str) – Field name to check for.
v (Any)
field_name (str)

Return type:

TypeGuard[Union[StructLike, BaseModelLike, AttrsLike]]

Returns:

bool

advanced_alchemy.service.is_schema_without_field(v, field_name)[source]¶

Check if a value is a msgspec Struct, Pydantic model, or attrs instance without a specific field.

SQLModel table=True models are excluded.

Parameters:

v¶ (Any) – Value to check.
field_name¶ (str) – Field name to check for.
v (Any)
field_name (str)

Return type:

TypeGuard[Union[StructLike, BaseModelLike, AttrsLike]]

Returns:

bool

advanced_alchemy.service.is_sqlmodel_table_model(v)[source]¶

Check if a value is a SQLModel table model instance or class.

Detects the dual nature of SQLModel table=True models: they are both Pydantic BaseModel subclasses AND SQLAlchemy-mapped models (have __mapper__).

This check fires BEFORE is_pydantic_model() to prevent SQLModel table instances from being misidentified as plain Pydantic schemas.

Parameters:

v¶ (Any) – Value to check (instance or class).
v (Any)

Returns:

True if v is a SQLModel table model, False otherwise.

Return type:

bool

advanced_alchemy.service.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:

TypeVar(ModelT, bound= base.ModelProtocol)

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

advanced_alchemy.service.schema_dump(data, exclude_unset=True)[source]¶

Overloads:

data (RowMapping), exclude_unset (bool) → dict[str, Any]
data (Row[Any]), exclude_unset (bool) → dict[str, Any]
data (DTODataLike[Any]), exclude_unset (bool) → dict[str, Any]
data (ModelT), exclude_unset (bool) → ModelT
data (Any), exclude_unset (bool) → dict[str, Any]

Parameters:

data (Union[dict[str, Any], ModelT, SupportedSchemaModel, DTODataLike[ModelT], RowMapping, Row[Any]])
exclude_unset (bool)

Return type:

Union[dict[str, Any], ModelT]

Dump a data object to a dictionary.

Parameters:

data¶ – dict[str, Any] | advanced_alchemy.base.ModelProtocol | msgspec.Struct | pydantic.BaseModel | attrs class | litestar.dto.data_structures.DTOData[ModelT] | sqlalchemy.RowMapping | sqlalchemy.engine.row.Row
exclude_unset¶ – bool Whether to exclude unset values.
data (Union[dict[str, Any], ModelT, SupportedSchemaModel, DTODataLike[ModelT], RowMapping, Row[Any]])
exclude_unset (bool)

Returns:

type: dict[str, Any], ModelProtocol]

Return type:

Union[