services

class advanced_alchemy.service.AttrsInstance[source]

Bases: Protocol

__init__(*args, **kwargs)
class advanced_alchemy.service.Empty[source]

Bases: object

A sentinel class used as placeholder.

class advanced_alchemy.service.ErrorMessages[source]

Bases: TypedDict

duplicate_key: Union[str, Callable[[Exception], str]]
integrity: Union[str, Callable[[Exception], str]]
foreign_key: Union[str, Callable[[Exception], str]]
multiple_rows: Union[str, Callable[[Exception], str]]
check_constraint: Union[str, Callable[[Exception], str]]
other: Union[str, Callable[[Exception], str]]
not_found: Union[str, Callable[[Exception], str]]
class advanced_alchemy.service.OffsetPagination[source]

Bases: Generic[T]

Container for data returned using limit/offset pagination.

items: Sequence[TypeVar(T)]

List of data being sent as part of the response.

limit: int

Maximal number of items to send.

offset: int

Offset from the beginning of the query.

Identical to an index.

total: int

Total number of items.

__init__(items, limit, offset, total)
Parameters:
Return type:

None

class advanced_alchemy.service.ResultConverter[source]

Bases: object

Simple mixin to help convert to a paginated response model.

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]

Convert the object to a response schema.

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

Parameters:
Raises:

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

Returns:

ModelProtocol | sqlalchemy.orm.RowMapping | OffsetPagination | msgspec.Struct | pydantic.BaseModel | attrs class

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

Return type:

AsyncIterator[Self]

Parameters:
class advanced_alchemy.service.SQLAlchemyAsyncRepositoryReadService[source]

Bases: ResultConverter, Generic[ModelT, SQLAlchemyAsyncRepositoryT]

Service object that operates on a repository object.

__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:
Return type:

None

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

Count of records returned by query.

Parameters:
Return type:

int

Returns:

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

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.

execution_options: ClassVar[Optional[dict[str, Any]]] = None

Default execution options for the repository.

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

Wrap repository exists operation.

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

Wrap repository scalar operation.

Parameters:

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

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

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

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

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

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

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

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

  • item_id (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)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

Representation of instance with identifier item_id.

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

Wrap repository scalar operation.

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

Wrap repository scalar operation.

Parameters:
Return type:

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

Returns:

Representation of instance with identifier item_id.

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

Wrap repository scalars operation.

Parameters:
Return type:

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

Returns:

The list of instances retrieved from the repository.

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, **kwargs)[source]

List of records and total count returned by query.

Parameters:
Return type:

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

Returns:

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

loader_options: ClassVar[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]] = None

Default loader options for the repository.

match_fields: ClassVar[Union[list[str], str, None]] = None

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

property model_type: type[ModelT][source]

Return the model type.

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.

Return type:

AsyncIterator[Self]

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)

property repository: SQLAlchemyAsyncRepositoryT

Return the repository instance.

Raises:

ImproperConfigurationError – If the repository is not initialized.

Returns:

The repository instance.

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

uniquify: ClassVar[bool] = False

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

repository_type: type[TypeVar(SQLAlchemyAsyncRepositoryT, bound= Union[SQLAlchemyAsyncRepository[Any], SQLAlchemyAsyncMockRepository[Any]])]

Type of the repository to use.

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)[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

  • data (ModelDictT[ModelT])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • auto_refresh (bool | None)

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

Returns:

Representation of created instance.

Return type:

ModelT

async create_many(data, *, auto_commit=None, auto_expunge=None, error_messages=Empty)[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

  • data (BulkModelDictT[ModelT])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

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

Returns:

Representation of created instances.

Return type:

Sequence[ModelT]

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

Wrap repository delete operation.

Parameters:

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

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

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

  • item_id (Any)

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

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

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

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

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

  • uniquify (bool | None)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

Representation of the deleted instance.

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

Wrap repository bulk instance deletion.

Parameters:

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

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

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

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

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

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

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

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

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

  • item_ids (list[Any])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

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

  • chunk_size (int | None)

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

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

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

  • uniquify (bool | None)

Return type:

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

Returns:

Representation of removed instances.

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

Wrap repository scalars operation.

Parameters:
Return type:

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

Returns:

The list of instances deleted from the repository.

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

  • **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)

  • kwargs (Any)

Return type:

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

Returns:

Representation of updated instance.

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

  • **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)

  • kwargs (Any)

Return type:

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

Returns:

Representation of created instance.

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)[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.

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

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)[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.

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

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)[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.

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

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)[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.

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

Returns:

Updated or created representation.

Return type:

Sequence[ModelT]

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

Return type:

Iterator[Self]

Parameters:
class advanced_alchemy.service.SQLAlchemySyncRepositoryReadService[source]

Bases: ResultConverter, Generic[ModelT, SQLAlchemySyncRepositoryT]

Service object that operates on a repository object.

__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:
Return type:

None

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

Count of records returned by query.

Parameters:
Return type:

int

Returns:

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

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.

execution_options: ClassVar[Optional[dict[str, Any]]] = None

Default execution options for the repository.

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

Wrap repository exists operation.

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

Wrap repository scalar operation.

Parameters:

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

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

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

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

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

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

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

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

  • item_id (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)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

Representation of instance with identifier item_id.

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

Wrap repository scalar operation.

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

Wrap repository scalar operation.

Parameters:
Return type:

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

Returns:

Representation of instance with identifier item_id.

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

Wrap repository scalars operation.

Parameters:
Return type:

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

Returns:

The list of instances retrieved from the repository.

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, **kwargs)[source]

List of records and total count returned by query.

Parameters:
Return type:

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

Returns:

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

loader_options: ClassVar[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]] = None

Default loader options for the repository.

match_fields: ClassVar[Union[list[str], str, None]] = None

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

property model_type: type[ModelT][source]

Return the model type.

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.

Return type:

Iterator[Self]

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)

property repository: SQLAlchemySyncRepositoryT

Return the repository instance.

Raises:

ImproperConfigurationError – If the repository is not initialized.

Returns:

The repository instance.

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

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

uniquify: ClassVar[bool] = False

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

repository_type: type[TypeVar(SQLAlchemySyncRepositoryT, bound= Union[SQLAlchemySyncRepository[Any], SQLAlchemySyncMockRepository[Any]])]

Type of the repository to use.

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)[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

  • data (ModelDictT[ModelT])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

  • auto_refresh (bool | None)

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

Returns:

Representation of created instance.

Return type:

ModelT

create_many(data, *, auto_commit=None, auto_expunge=None, error_messages=Empty)[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

  • data (BulkModelDictT[ModelT])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

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

Returns:

Representation of created instances.

Return type:

Sequence[ModelT]

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

Wrap repository delete operation.

Parameters:

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

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

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

  • item_id (Any)

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

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

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

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

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

  • uniquify (bool | None)

Return type:

TypeVar(ModelT, bound= base.ModelProtocol)

Returns:

Representation of the deleted instance.

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

Wrap repository bulk instance deletion.

Parameters:

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

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

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

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

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

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

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

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

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

  • item_ids (list[Any])

  • auto_commit (bool | None)

  • auto_expunge (bool | None)

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

  • chunk_size (int | None)

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

  • load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)

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

  • uniquify (bool | None)

Return type:

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

Returns:

Representation of removed instances.

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

Wrap repository scalars operation.

Parameters:
Return type:

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

Returns:

The list of instances deleted from the repository.

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

  • **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)

  • kwargs (Any)

Return type:

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

Returns:

Representation of updated instance.

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

  • **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)

  • kwargs (Any)

Return type:

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

Returns:

Representation of created instance.

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)[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.

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

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)[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.

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

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)[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.

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

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)[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.

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

Returns:

Updated or created representation.

Return type:

Sequence[ModelT]

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) – Class to introspect.

Raises:
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:
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:
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:
Return type:

TypeGuard[DTODataLike[Any]]

Returns:

bool

advanced_alchemy.service.is_msgspec_struct(v)[source]

Check if a value is a msgspec struct.

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

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

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.

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.model_from_dict(model, **kwargs)[source]

Create an ORM model instance from a dictionary of attributes.

Parameters:

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

  • **kwargs (Any) – Keyword arguments containing model attribute values.

  • model (type[ModelT])

  • kwargs (Any)

Returns:

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

Return type:

ModelT

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

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[