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: ModelOrRowMappingT, total: int | None = None, filters: Sequence[StatementFilter | ColumnElement[bool]] | Sequence[StatementFilter] | None = None, *, schema_type: None = None) → ModelOrRowMappingT[source]¶

to_schema(data: Sequence[advanced_alchemy.repository.ModelOrRowMappingT], total: int | None = None, filters: Sequence[StatementFilter | ColumnElement[bool]] | Sequence[StatementFilter] | None = None, *, schema_type: None = None) → OffsetPagination[ModelOrRowMappingT]

to_schema(data: Sequence[ModelProtocol] | Sequence[RowMapping], total: int | None = None, filters: Sequence[StatementFilter | ColumnElement[bool]] | Sequence[StatementFilter] | None = None, *, schema_type: type[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

Returns:

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

class advanced_alchemy.service.SQLAlchemyAsyncQueryService[source]¶

Bases: ResultConverter

Simple service to execute the basic Query repository..

__init__(session: AsyncSession | async_scoped_session[AsyncSession], **repo_kwargs: Any) → None[source]¶

Configure the service object.

Parameters:

session¶ – Session managing the unit-of-work for the operation.
**repo_kwargs¶ – Optional configuration values to pass into the repository

classmethod new(session: AsyncSession | async_scoped_session[AsyncSession] | None = None, config: SQLAlchemyAsyncConfig | None = None) → AsyncIterator[Self][source]¶

Context manager that returns instance of service object.

Handles construction of the database session._create_select_for_model

Returns:: The service object instance.

class advanced_alchemy.service.SQLAlchemyAsyncRepositoryReadService[source]¶

Bases: ResultConverter, Generic[ModelT]

Service object that operates on a repository object.

repository_type: type[SQLAlchemyAsyncRepositoryProtocol[ModelT] | SQLAlchemyAsyncSlugRepositoryProtocol[ModelT]]¶: Type of the repository to use.

loader_options: LoadSpec | None = None¶: Default loader options for the repository.

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

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

__init__(session: AsyncSession | async_scoped_session[AsyncSession], statement: Select[tuple[ModelT]] | None = None, auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: list[OrderingPair] | OrderingPair | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None, **repo_kwargs: Any) → None[source]¶

Configure the service object.

Parameters:

session¶ – Session managing the unit-of-work for the operation.
statement¶ – To facilitate customization of the underlying select query.
auto_expunge¶ – Remove object from session before returning.
auto_refresh¶ – Refresh object from session before returning.
auto_commit¶ – Commit objects before returning.
order_by¶ – Set default order options for queries.
error_messages¶ – A set of custom error messages to use for operations
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
**repo_kwargs¶ – passed as keyword args to repo instantiation.

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

Count of records returned by query.

Parameters:

*filters¶ – arguments for filtering.
statement¶ – To facilitate customization of the underlying select query.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set relationships to be loaded
execution_options¶ – Set default execution options
**kwargs¶ – key value pairs of filter types.

Returns:

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

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

Wrap repository exists operation.

Parameters:

*filters¶ – Types for specific filtering operations.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set relationships to be loaded
execution_options¶ – Set default execution options
**kwargs¶ – Keyword arguments for attribute based filtering.

Returns:

Representation of instance with identifier item_id.

Wrap repository scalar operation.

Parameters:

item_id¶ – Identifier of instance to be retrieved.
auto_expunge¶ – Remove object from session before returning.
statement¶ – To facilitate customization of the underlying select query.
id_attribute¶ – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set relationships to be loaded
execution_options¶ – Set default execution options

Returns:

Representation of instance with identifier item_id.

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

Wrap repository scalar operation.

Parameters:

*filters¶ – Types for specific filtering operations.
auto_expunge¶ – Remove object from session before returning.
statement¶ – To facilitate customization of the underlying select query.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
**kwargs¶ – Identifier of the instance to be retrieved.

Returns:

Representation of instance with identifier item_id.

Wrap repository scalar operation.

Parameters:

*filters¶ – Types for specific filtering operations.
auto_expunge¶ – Remove object from session before returning.
statement¶ – To facilitate customization of the underlying select query.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
**kwargs¶ – Identifier of the instance to be retrieved.

Returns:

Representation of instance with identifier item_id.

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

Returns:

Representation of created instances.

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

List of records and total count returned by query.

Parameters:

*filters¶ – Types for specific filtering operations.
statement¶ – To facilitate customization of the underlying select query.
auto_expunge¶ – Remove object from session before returning.
force_basic_query_mode¶ – Force list and count to use two queries instead of an analytical window function.
order_by¶ – Set default order options for queries.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set relationships to be loaded
execution_options¶ – Set default execution options
**kwargs¶ – Instance attribute value filters.

Returns:

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

classmethod new(session: AsyncSession | async_scoped_session[AsyncSession] | None = None, statement: Select[tuple[ModelT]] | None = None, config: SQLAlchemyAsyncConfig | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) → AsyncIterator[Self][source]¶

Context manager that returns instance of service object.

Handles construction of the database session._create_select_for_model

Returns:: The service object instance.

Wrap repository scalars operation.

Parameters:

*filters¶ – Types for specific filtering operations.
auto_expunge¶ – Remove object from session before returning.
statement¶ – To facilitate customization of the underlying select query.
order_by¶ – Set default order options for queries.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
**kwargs¶ – Instance attribute value filters.

Returns:

The list of instances retrieved from the repository.

class advanced_alchemy.service.SQLAlchemyAsyncRepositoryService[source]¶

Bases: SQLAlchemyAsyncRepositoryReadService[ModelT]

Service object that operates on a repository object.

async create(data: ~typing.Dict[str, ~typing.Any] | advanced_alchemy.repository.typing.ModelT | ~msgspec.Struct | ~pydantic.main.BaseModel | ~litestar.dto.data_structures.DTOData[advanced_alchemy.repository.typing.ModelT], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, auto_refresh: bool | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>) → advanced_alchemy.repository.typing.ModelT[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

Returns:

Representation of created instance.

async create_many(data: ~typing.Sequence[~typing.Dict[str, ~typing.Any] | advanced_alchemy.repository.typing.ModelT | ~msgspec.Struct | ~pydantic.main.BaseModel] | ~litestar.dto.data_structures.DTOData[~typing.List[advanced_alchemy.repository.typing.ModelT]], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>) → Sequence[advanced_alchemy.repository.typing.ModelT][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

Returns:

Representation of created instances.

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

Returns:

Updated representation.

async update_many(data: ~typing.Sequence[~typing.Dict[str, ~typing.Any] | advanced_alchemy.repository.typing.ModelT | ~msgspec.Struct | ~pydantic.main.BaseModel] | ~litestar.dto.data_structures.DTOData[~typing.List[advanced_alchemy.repository.typing.ModelT]], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None) → Sequence[advanced_alchemy.repository.typing.ModelT][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

Returns:

Representation of updated instances.

async upsert(data: ~typing.Dict[str, ~typing.Any] | advanced_alchemy.repository.typing.ModelT | ~msgspec.Struct | ~pydantic.main.BaseModel | ~litestar.dto.data_structures.DTOData[advanced_alchemy.repository.typing.ModelT], item_id: ~typing.Any | None = None, *, attribute_names: ~typing.Iterable[str] | None = None, with_for_update: bool | None = None, auto_expunge: bool | None = None, auto_commit: bool | None = None, auto_refresh: bool | None = None, match_fields: list[str] | str | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None) → advanced_alchemy.repository.typing.ModelT[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

Returns:

Updated or created representation.

async upsert_many(data: ~typing.Sequence[~typing.Dict[str, ~typing.Any] | advanced_alchemy.repository.typing.ModelT | ~msgspec.Struct | ~pydantic.main.BaseModel] | ~litestar.dto.data_structures.DTOData[~typing.List[advanced_alchemy.repository.typing.ModelT]], *, auto_expunge: bool | None = None, auto_commit: bool | None = None, no_merge: bool = False, match_fields: list[str] | str | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None) → Sequence[advanced_alchemy.repository.typing.ModelT][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

Returns:

Updated or created representation.

Wrap repository instance creation.

Parameters:

*filters¶ – Types for specific filtering operations.
match_fields¶ – a list of keys to use to match the existing model. When empty, all fields are matched.
upsert¶ – When using match_fields and actual model values differ from kwargs, perform an update operation on the model.
create¶ – Should a model be created. If no model is found, an exception is raised.
attribute_names¶ – an iterable of attribute names to pass into the update method.
with_for_update¶ – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT
auto_expunge¶ – Remove object from session before returning.
auto_refresh¶ – Refresh object from session before returning.
auto_commit¶ – Commit objects before returning.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
**kwargs¶ – Identifier of the instance to be retrieved.

Returns:

Representation of created instance.

Wrap repository instance creation.

Parameters:

*filters¶ – Types for specific filtering operations.
match_fields¶ – a list of keys to use to match the existing model. When empty, all fields are matched.
attribute_names¶ – an iterable of attribute names to pass into the update method.
with_for_update¶ – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT
auto_expunge¶ – Remove object from session before returning.
auto_refresh¶ – Refresh object from session before returning.
auto_commit¶ – Commit objects before returning.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
**kwargs¶ – Identifier of the instance to be retrieved.

Returns:

Representation of updated instance.

Wrap repository delete operation.

Parameters:

item_id¶ – Identifier of instance to be deleted.
auto_expunge¶ – Remove object from session before returning.
auto_commit¶ – Commit objects before returning.
id_attribute¶ – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options

Returns:

Representation of the deleted instance.

Wrap repository bulk instance deletion.

Parameters:

item_ids¶ – Identifier of instance to be deleted.
auto_expunge¶ – Remove object from session before returning.
auto_commit¶ – Commit objects before returning.
id_attribute¶ – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.
chunk_size¶ – Allows customization of the insertmanyvalues_max_parameters setting for the driver. Defaults to 950 if left unset.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options

Returns:

Representation of removed instances.

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

Wrap repository scalars operation.

Parameters:

*filters¶ – Types for specific filtering operations.
auto_expunge¶ – Remove object from session before returning.
auto_commit¶ – Commit objects before returning.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
sanity_check¶ – When true, the length of selected instances is compared to the deleted row count
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
**kwargs¶ – Instance attribute value filters.

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: Session | scoped_session[Session], **repo_kwargs: Any) → None[source]¶

Configure the service object.

Parameters:

session¶ – Session managing the unit-of-work for the operation.
**repo_kwargs¶ – Optional configuration values to pass into the repository

classmethod new(session: Session | scoped_session[Session] | None = None, config: SQLAlchemySyncConfig | None = None) → Iterator[Self][source]¶

Context manager that returns instance of service object.

Handles construction of the database session._create_select_for_model

Returns:: The service object instance.

class advanced_alchemy.service.SQLAlchemySyncRepositoryReadService[source]¶

Bases: ResultConverter, Generic[ModelT]

Service object that operates on a repository object.

repository_type: type[SQLAlchemySyncRepositoryProtocol[ModelT] | SQLAlchemySyncSlugRepositoryProtocol[ModelT]]¶: Type of the repository to use.

loader_options: LoadSpec | None = None¶: Default loader options for the repository.

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

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

__init__(session: Session | scoped_session[Session], statement: Select[tuple[ModelT]] | None = None, auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: list[OrderingPair] | OrderingPair | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None, **repo_kwargs: Any) → None[source]¶

Configure the service object.

Parameters:

session¶ – Session managing the unit-of-work for the operation.
statement¶ – To facilitate customization of the underlying select query.
auto_expunge¶ – Remove object from session before returning.
auto_refresh¶ – Refresh object from session before returning.
auto_commit¶ – Commit objects before returning.
order_by¶ – Set default order options for queries.
error_messages¶ – A set of custom error messages to use for operations
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
**repo_kwargs¶ – passed as keyword args to repo instantiation.

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

Count of records returned by query.

Parameters:

*filters¶ – arguments for filtering.
statement¶ – To facilitate customization of the underlying select query.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set relationships to be loaded
execution_options¶ – Set default execution options
**kwargs¶ – key value pairs of filter types.

Returns:

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

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

Wrap repository exists operation.

Parameters:

*filters¶ – Types for specific filtering operations.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set relationships to be loaded
execution_options¶ – Set default execution options
**kwargs¶ – Keyword arguments for attribute based filtering.

Returns:

Representation of instance with identifier item_id.

Wrap repository scalar operation.

Parameters:

item_id¶ – Identifier of instance to be retrieved.
auto_expunge¶ – Remove object from session before returning.
statement¶ – To facilitate customization of the underlying select query.
id_attribute¶ – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set relationships to be loaded
execution_options¶ – Set default execution options

Returns:

Representation of instance with identifier item_id.

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

Wrap repository scalar operation.

Parameters:

*filters¶ – Types for specific filtering operations.
auto_expunge¶ – Remove object from session before returning.
statement¶ – To facilitate customization of the underlying select query.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
**kwargs¶ – Identifier of the instance to be retrieved.

Returns:

Representation of instance with identifier item_id.

Wrap repository scalar operation.

Parameters:

*filters¶ – Types for specific filtering operations.
auto_expunge¶ – Remove object from session before returning.
statement¶ – To facilitate customization of the underlying select query.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
**kwargs¶ – Identifier of the instance to be retrieved.

Returns:

Representation of instance with identifier item_id.

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

Returns:

Representation of created instances.

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

List of records and total count returned by query.

Parameters:

*filters¶ – Types for specific filtering operations.
statement¶ – To facilitate customization of the underlying select query.
auto_expunge¶ – Remove object from session before returning.
force_basic_query_mode¶ – Force list and count to use two queries instead of an analytical window function.
order_by¶ – Set default order options for queries.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set relationships to be loaded
execution_options¶ – Set default execution options
**kwargs¶ – Instance attribute value filters.

Returns:

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

Returns:: The service object instance.

Wrap repository scalars operation.

Parameters:

*filters¶ – Types for specific filtering operations.
auto_expunge¶ – Remove object from session before returning.
statement¶ – To facilitate customization of the underlying select query.
order_by¶ – Set default order options for queries.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
**kwargs¶ – Instance attribute value filters.

Returns:

The list of instances retrieved from the repository.

class advanced_alchemy.service.SQLAlchemySyncRepositoryService[source]¶

Bases: SQLAlchemySyncRepositoryReadService[ModelT]

Service object that operates on a repository object.

create(data: ~typing.Dict[str, ~typing.Any] | advanced_alchemy.repository.typing.ModelT | ~msgspec.Struct | ~pydantic.main.BaseModel | ~litestar.dto.data_structures.DTOData[advanced_alchemy.repository.typing.ModelT], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, auto_refresh: bool | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>) → advanced_alchemy.repository.typing.ModelT[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

Returns:

Representation of created instance.

create_many(data: ~typing.Sequence[~typing.Dict[str, ~typing.Any] | advanced_alchemy.repository.typing.ModelT | ~msgspec.Struct | ~pydantic.main.BaseModel] | ~litestar.dto.data_structures.DTOData[~typing.List[advanced_alchemy.repository.typing.ModelT]], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>) → Sequence[advanced_alchemy.repository.typing.ModelT][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

Returns:

Representation of created instances.

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

Returns:

Updated representation.

update_many(data: ~typing.Sequence[~typing.Dict[str, ~typing.Any] | advanced_alchemy.repository.typing.ModelT | ~msgspec.Struct | ~pydantic.main.BaseModel] | ~litestar.dto.data_structures.DTOData[~typing.List[advanced_alchemy.repository.typing.ModelT]], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None) → Sequence[advanced_alchemy.repository.typing.ModelT][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

Returns:

Representation of updated instances.

upsert(data: ~typing.Dict[str, ~typing.Any] | advanced_alchemy.repository.typing.ModelT | ~msgspec.Struct | ~pydantic.main.BaseModel | ~litestar.dto.data_structures.DTOData[advanced_alchemy.repository.typing.ModelT], item_id: ~typing.Any | None = None, *, attribute_names: ~typing.Iterable[str] | None = None, with_for_update: bool | None = None, auto_expunge: bool | None = None, auto_commit: bool | None = None, auto_refresh: bool | None = None, match_fields: list[str] | str | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None) → advanced_alchemy.repository.typing.ModelT[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

Returns:

Updated or created representation.

upsert_many(data: ~typing.Sequence[~typing.Dict[str, ~typing.Any] | advanced_alchemy.repository.typing.ModelT | ~msgspec.Struct | ~pydantic.main.BaseModel] | ~litestar.dto.data_structures.DTOData[~typing.List[advanced_alchemy.repository.typing.ModelT]], *, auto_expunge: bool | None = None, auto_commit: bool | None = None, no_merge: bool = False, match_fields: list[str] | str | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | None | ~typing.Type[~advanced_alchemy.utils.dataclass.Empty] = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~typing.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~typing.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None) → Sequence[advanced_alchemy.repository.typing.ModelT][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

Returns:

Updated or created representation.

Wrap repository instance creation.

Parameters:

*filters¶ – Types for specific filtering operations.
match_fields¶ – a list of keys to use to match the existing model. When empty, all fields are matched.
upsert¶ – When using match_fields and actual model values differ from kwargs, perform an update operation on the model.
create¶ – Should a model be created. If no model is found, an exception is raised.
attribute_names¶ – an iterable of attribute names to pass into the update method.
with_for_update¶ – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT
auto_expunge¶ – Remove object from session before returning.
auto_refresh¶ – Refresh object from session before returning.
auto_commit¶ – Commit objects before returning.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
**kwargs¶ – Identifier of the instance to be retrieved.

Returns:

Representation of created instance.

Wrap repository instance creation.

Parameters:

*filters¶ – Types for specific filtering operations.
match_fields¶ – a list of keys to use to match the existing model. When empty, all fields are matched.
attribute_names¶ – an iterable of attribute names to pass into the update method.
with_for_update¶ – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT
auto_expunge¶ – Remove object from session before returning.
auto_refresh¶ – Refresh object from session before returning.
auto_commit¶ – Commit objects before returning.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
**kwargs¶ – Identifier of the instance to be retrieved.

Returns:

Representation of updated instance.

Wrap repository delete operation.

Parameters:

item_id¶ – Identifier of instance to be deleted.
auto_expunge¶ – Remove object from session before returning.
auto_commit¶ – Commit objects before returning.
id_attribute¶ – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options

Returns:

Representation of the deleted instance.

Wrap repository bulk instance deletion.

Parameters:

item_ids¶ – Identifier of instance to be deleted.
auto_expunge¶ – Remove object from session before returning.
auto_commit¶ – Commit objects before returning.
id_attribute¶ – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.
chunk_size¶ – Allows customization of the insertmanyvalues_max_parameters setting for the driver. Defaults to 950 if left unset.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options

Returns:

Representation of removed instances.

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

Wrap repository scalars operation.

Parameters:

*filters¶ – Types for specific filtering operations.
auto_expunge¶ – Remove object from session before returning.
auto_commit¶ – Commit objects before returning.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
sanity_check¶ – When true, the length of selected instances is compared to the deleted row count
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
**kwargs¶ – Instance attribute value filters.

Returns:

The list of instances deleted from the repository.

advanced_alchemy.service.find_filter(filter_type: type[FilterTypeT], filters: Sequence[StatementFilter | ColumnElement[bool]] | Sequence[StatementFilter]) → FilterTypeT | None[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

Returns:

The match filter instance or None

advanced_alchemy.service.is_dict(v: Any) → TypeGuard[dict[str, Any]][source]¶

Check if a value is a dictionary.

Parameters:: v¶ – Value to check.
Returns:: bool

advanced_alchemy.service.is_dict_with_field(v: Any, field_name: str) → TypeGuard[dict[str, Any]][source]¶

Check if a dictionary has a specific field.

Parameters:

v¶ – Value to check.
field_name¶ – Field name to check for.

Returns:

bool

advanced_alchemy.service.is_dict_without_field(v: Any, field_name: str) → TypeGuard[dict[str, Any]][source]¶

Check if a dictionary does not have a specific field.

Parameters:

v¶ – Value to check.
field_name¶ – Field name to check for.

Returns:

bool

advanced_alchemy.service.is_msgspec_model(v: Any) → TypeGuard[msgspec.Struct][source]¶

Check if a value is a msgspec model.

Parameters:: v¶ – Value to check.
Returns:: bool

advanced_alchemy.service.is_msgspec_model_with_field(v: Any, field_name: str) → TypeGuard[msgspec.Struct][source]¶

Check if a msgspec model has a specific field.

Parameters:

v¶ – Value to check.
field_name¶ – Field name to check for.

Returns:

bool

advanced_alchemy.service.is_msgspec_model_without_field(v: Any, field_name: str) → TypeGuard[msgspec.Struct][source]¶

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

Parameters:

v¶ – Value to check.
field_name¶ – Field name to check for.

Returns:

bool

advanced_alchemy.service.is_pydantic_model(v: Any) → TypeGuard[pydantic.BaseModel][source]¶

Check if a value is a pydantic model.

Parameters:: v¶ – Value to check.
Returns:: bool

advanced_alchemy.service.is_pydantic_model_with_field(v: Any, field_name: str) → TypeGuard[pydantic.BaseModel][source]¶

Check if a pydantic model has a specific field.

Parameters:

v¶ – Value to check.
field_name¶ – Field name to check for.

Returns:

bool

advanced_alchemy.service.is_pydantic_model_without_field(v: Any, field_name: str) → TypeGuard[pydantic.BaseModel][source]¶

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

Parameters:

v¶ – Value to check.
field_name¶ – Field name to check for.

Returns:

bool

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

Create an ORM model instance from a dictionary of attributes.

Parameters:

model¶ – The SQLAlchemy model class to instantiate.
**kwargs¶ – Keyword arguments containing model attribute values.

Returns:

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

Return type:

ModelT

Dump a data object to a dictionary.

Parameters:

data¶ – dict[str, Any] | ModelProtocol | msgspec.Struct | pydantic.BaseModel | litestar.dto.data_structures.DTOData[ModelT]
exclude_unset¶ – bool Whether to exclude unset values.

Returns:

type: dict[str, Any], ModelProtocol]

Return type:

Union[