services¶
- 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'>) ModelT [source]¶
Wrap repository instance creation.
- Parameters:
data¶ – Representation to be created.
auto_expunge¶ – Remove object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.auto_refresh¶ – Refresh object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_refresh
auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
- Returns:
Representation of created instance.
- async create_many(data: BulkModelDictT[ModelT], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>) Sequence[ModelT] [source]¶
Wrap repository bulk instance creation.
- Parameters:
data¶ – Representations to be created.
auto_expunge¶ – Remove object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
- Returns:
Representation of created instances.
- async update(data: ModelDictT[ModelT], item_id: Any | None = None, *, attribute_names: Iterable[str] | None = None, with_for_update: bool | None = None, auto_commit: bool | None = None, auto_expunge: bool | None = None, auto_refresh: bool | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) ModelT [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. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.auto_refresh¶ – Refresh object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_refresh
auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
id_attribute¶ – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.
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: BulkModelDictT[ModelT], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) Sequence[ModelT] [source]¶
Wrap repository bulk instance update.
- Parameters:
data¶ – Representations to be updated.
auto_expunge¶ – Remove object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
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) 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. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.auto_refresh¶ – Refresh object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_refresh
auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
match_fields¶ – a list of keys to use to match the existing model. When empty, all fields are matched.
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: BulkModelDictT[ModelT], *, auto_expunge: bool | None = None, auto_commit: bool | None = None, no_merge: bool = False, match_fields: list[str] | str | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) Sequence[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 ofid_attribute
.auto_expunge¶ – Remove object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
no_merge¶ – Skip the usage of optimized Merge statements
SQLAlchemyAsyncRepository.auto_commit
match_fields¶ – a list of keys to use to match the existing model. When empty, 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 get_or_upsert(*filters: StatementFilter | ColumnElement[bool], match_fields: list[str] | str | None = None, upsert: bool = True, attribute_names: Iterable[str] | None = None, with_for_update: bool | None = None, auto_commit: bool | None = None, auto_expunge: bool | None = None, auto_refresh: bool | None = None, 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[ModelT, bool] [source]¶
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. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.auto_refresh¶ – Refresh object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_refresh
auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
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.
- async get_and_update(*filters: StatementFilter | ColumnElement[bool], match_fields: list[str] | str | None = None, attribute_names: Iterable[str] | None = None, with_for_update: bool | None = None, auto_commit: bool | None = None, auto_expunge: bool | None = None, auto_refresh: bool | None = None, 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[ModelT, bool] [source]¶
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. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.auto_refresh¶ – Refresh object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_refresh
auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
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.
- async delete(item_id: Any, *, auto_commit: bool | None = None, auto_expunge: bool | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) ModelT [source]¶
Wrap repository delete operation.
- Parameters:
item_id¶ – Identifier of instance to be deleted.
auto_expunge¶ – Remove object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
id_attribute¶ – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.
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.
- async delete_many(item_ids: list[Any], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, chunk_size: int | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) Sequence[ModelT] [source]¶
Wrap repository bulk instance deletion.
- Parameters:
item_ids¶ – Identifier of instance to be deleted.
auto_expunge¶ – Remove object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
id_attribute¶ – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.
chunk_size¶ – Allows customization of the
insertmanyvalues_max_parameters
setting for the driver. Defaults to 950 if left unset.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. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
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.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.
- 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.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.
- 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:
Generic
[ModelT
],ResultConverter
Service object that operates on a repository object.
- __init__(session: Session | scoped_session[Session], statement: Select[tuple[ModelT]] | StatementLambdaElement | 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]] | StatementLambdaElement | 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. Defaults to
SQLAlchemyAsyncRepository.statement
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.
- get(item_id: Any, *, statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, auto_expunge: bool | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) ModelT [source]¶
Wrap repository scalar operation.
- Parameters:
item_id¶ – Identifier of instance to be retrieved.
auto_expunge¶ – Remove object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
statement¶ – To facilitate customization of the underlying select query. Defaults to
SQLAlchemyAsyncRepository.statement
id_attribute¶ – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.
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]] | StatementLambdaElement | 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. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
statement¶ – To facilitate customization of the underlying select query. Defaults to
SQLAlchemyAsyncRepository.statement
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.
- get_one_or_none(*filters: StatementFilter | ColumnElement[bool], statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, auto_expunge: bool | 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) ModelT | None [source]¶
Wrap repository scalar operation.
- Parameters:
*filters¶ – Types for specific filtering operations.
auto_expunge¶ – Remove object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
statement¶ – To facilitate customization of the underlying select query. Defaults to
SQLAlchemyAsyncRepository.statement
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.
- to_model(data: Dict[str, Any] | ModelT | Struct | BaseModel | DTOData[ModelT], operation: str | None = None) ModelT [source]¶
Parse and Convert input into a model.
- list_and_count(*filters: StatementFilter | ColumnElement[bool], statement: Select[tuple[ModelT]] | StatementLambdaElement | 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. Defaults to
SQLAlchemyAsyncRepository.statement
auto_expunge¶ – Remove object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.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: Session | scoped_session[Session] | None = None, statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, config: SQLAlchemySyncConfig | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | 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.
- list(*filters: StatementFilter | ColumnElement[bool], statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, auto_expunge: 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) Sequence[ModelT] [source]¶
Wrap repository scalars operation.
- Parameters:
*filters¶ – Types for specific filtering operations.
auto_expunge¶ – Remove object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
statement¶ – To facilitate customization of the underlying select query. Defaults to
SQLAlchemyAsyncRepository.statement
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'>) ModelT [source]¶
Wrap repository instance creation.
- Parameters:
data¶ – Representation to be created.
auto_expunge¶ – Remove object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.auto_refresh¶ – Refresh object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_refresh
auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
- Returns:
Representation of created instance.
- create_many(data: BulkModelDictT[ModelT], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>) Sequence[ModelT] [source]¶
Wrap repository bulk instance creation.
- Parameters:
data¶ – Representations to be created.
auto_expunge¶ – Remove object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
- Returns:
Representation of created instances.
- update(data: ModelDictT[ModelT], item_id: Any | None = None, *, attribute_names: Iterable[str] | None = None, with_for_update: bool | None = None, auto_commit: bool | None = None, auto_expunge: bool | None = None, auto_refresh: bool | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) ModelT [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. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.auto_refresh¶ – Refresh object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_refresh
auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
id_attribute¶ – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.
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: BulkModelDictT[ModelT], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) Sequence[ModelT] [source]¶
Wrap repository bulk instance update.
- Parameters:
data¶ – Representations to be updated.
auto_expunge¶ – Remove object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
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) 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. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.auto_refresh¶ – Refresh object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_refresh
auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
match_fields¶ – a list of keys to use to match the existing model. When empty, all fields are matched.
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: BulkModelDictT[ModelT], *, auto_expunge: bool | None = None, auto_commit: bool | None = None, no_merge: bool = False, match_fields: list[str] | str | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) Sequence[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 ofid_attribute
.auto_expunge¶ – Remove object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
no_merge¶ – Skip the usage of optimized Merge statements
SQLAlchemyAsyncRepository.auto_commit
match_fields¶ – a list of keys to use to match the existing model. When empty, 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.
- get_or_upsert(*filters: StatementFilter | ColumnElement[bool], match_fields: list[str] | str | None = None, upsert: bool = True, attribute_names: Iterable[str] | None = None, with_for_update: bool | None = None, auto_commit: bool | None = None, auto_expunge: bool | None = None, auto_refresh: bool | None = None, 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[ModelT, bool] [source]¶
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. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.auto_refresh¶ – Refresh object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_refresh
auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
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.
- get_and_update(*filters: StatementFilter | ColumnElement[bool], match_fields: list[str] | str | None = None, attribute_names: Iterable[str] | None = None, with_for_update: bool | None = None, auto_commit: bool | None = None, auto_expunge: bool | None = None, auto_refresh: bool | None = None, 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[ModelT, bool] [source]¶
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. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.auto_refresh¶ – Refresh object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_refresh
auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
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.
- delete(item_id: Any, *, auto_commit: bool | None = None, auto_expunge: bool | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) ModelT [source]¶
Wrap repository delete operation.
- Parameters:
item_id¶ – Identifier of instance to be deleted.
auto_expunge¶ – Remove object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
id_attribute¶ – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.
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.
- delete_many(item_ids: list[Any], *, auto_commit: bool | None = None, auto_expunge: bool | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, chunk_size: int | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) Sequence[ModelT] [source]¶
Wrap repository bulk instance deletion.
- Parameters:
item_ids¶ – Identifier of instance to be deleted.
auto_expunge¶ – Remove object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
id_attribute¶ – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.
chunk_size¶ – Allows customization of the
insertmanyvalues_max_parameters
setting for the driver. Defaults to 950 if left unset.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. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
auto_commit¶ – Commit objects before returning. Defaults to
SQLAlchemyAsyncRepository.auto_commit
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.SQLAlchemyAsyncRepositoryReadService[source]¶
Bases:
Generic
[ModelT
],ResultConverter
Service object that operates on a repository object.
- __init__(session: AsyncSession | async_scoped_session[AsyncSession], statement: Select[tuple[ModelT]] | StatementLambdaElement | 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]] | StatementLambdaElement | 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. Defaults to
SQLAlchemyAsyncRepository.statement
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.
- async get(item_id: Any, *, statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, id_attribute: str | InstrumentedAttribute[Any] | None = None, auto_expunge: bool | None = None, error_messages: ErrorMessages | None | EmptyType = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: LoadSpec | None = None, execution_options: dict[str, Any] | None = None) ModelT [source]¶
Wrap repository scalar operation.
- Parameters:
item_id¶ – Identifier of instance to be retrieved.
auto_expunge¶ – Remove object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
statement¶ – To facilitate customization of the underlying select query. Defaults to
SQLAlchemyAsyncRepository.statement
id_attribute¶ – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.
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]] | StatementLambdaElement | 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. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
statement¶ – To facilitate customization of the underlying select query. Defaults to
SQLAlchemyAsyncRepository.statement
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.
- async get_one_or_none(*filters: StatementFilter | ColumnElement[bool], statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, auto_expunge: bool | 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) ModelT | None [source]¶
Wrap repository scalar operation.
- Parameters:
*filters¶ – Types for specific filtering operations.
auto_expunge¶ – Remove object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
statement¶ – To facilitate customization of the underlying select query. Defaults to
SQLAlchemyAsyncRepository.statement
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.
- async to_model(data: Dict[str, Any] | ModelT | Struct | BaseModel | DTOData[ModelT], operation: str | None = None) ModelT [source]¶
Parse and Convert input into a model.
- async list_and_count(*filters: StatementFilter | ColumnElement[bool], statement: Select[tuple[ModelT]] | StatementLambdaElement | 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. Defaults to
SQLAlchemyAsyncRepository.statement
auto_expunge¶ – Remove object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
.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]] | StatementLambdaElement | 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.
- async list(*filters: StatementFilter | ColumnElement[bool], statement: Select[tuple[ModelT]] | StatementLambdaElement | None = None, auto_expunge: 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) Sequence[ModelT] [source]¶
Wrap repository scalars operation.
- Parameters:
*filters¶ – Types for specific filtering operations.
auto_expunge¶ – Remove object from session before returning. Defaults to
SQLAlchemyAsyncRepository.auto_expunge
statement¶ – To facilitate customization of the underlying select query. Defaults to
SQLAlchemyAsyncRepository.statement
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.OffsetPagination[source]¶
Bases:
Generic
[T
]Container for data returned using limit/offset pagination.
- 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.
- class advanced_alchemy.service.ResultConverter[source]¶
Bases:
object
Simple mixin to help convert to a paginated response model the results set is a list.
- 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[ModelOrRowMappingT], total: int | None = None, filters: Sequence[StatementFilter | ColumnElement[bool]] | Sequence[StatementFilter] | None = None, *, schema_type: None = None) OffsetPagination[ModelOrRowMappingT]
- to_schema(data: ModelProtocol | RowMapping, total: int | None = None, filters: Sequence[StatementFilter | ColumnElement[bool]] | Sequence[StatementFilter] | None = None, *, schema_type: type[ModelDTOT]) ModelDTOT
- 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.
- advanced_alchemy.service.model_from_dict(model: type[ModelT], **kwargs: Any) ModelT [source]¶
Return ORM Object from Dictionary.