services¶
- class advanced_alchemy.service.OffsetPagination[source]¶
Bases:
Generic
[T
]Container for data returned using limit/offset pagination.
- class advanced_alchemy.service.ResultConverter[source]¶
Bases:
object
Simple mixin to help convert to a paginated response model.
Single objects are transformed to the supplied schema type, and lists of objects are automatically transformed into an OffsetPagination response of the supplied schema type.
- Parameters:
data¶ – A database model instance or row mapping. Type:
ModelOrRowMappingT
- Returns:
The converted schema object.
- to_schema(data, total=None, filters=None, *, schema_type=None)[source]¶
Convert the object to a response schema.
When schema_type is None, the model is returned with no conversion.
- Parameters:
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 todata (Union[ModelOrRowMappingT, Sequence[ModelOrRowMappingT], ModelProtocol, Sequence[ModelProtocol], RowMapping, Sequence[RowMapping], Row[Any], Sequence[Row[Any]], dict[str, Any], Sequence[dict[str, Any]]])
total (Optional[int])
filters (Union[Sequence[Union[StatementFilter, ColumnElement[bool]]], Sequence[StatementFilter], None])
schema_type (Optional[type[ModelDTOT]])
- Raises:
AdvancedAlchemyError – If schema_type is not a valid Pydantic, Msgspec, or attrs schema and all libraries are not installed.
- Returns:
ModelProtocol
|sqlalchemy.orm.RowMapping
|OffsetPagination
|msgspec.Struct
|pydantic.BaseModel
|attrs class
- Return type:
Union[ModelOrRowMappingT, OffsetPagination[ModelOrRowMappingT], ModelDTOT, OffsetPagination[ModelDTOT]]
- class advanced_alchemy.service.SQLAlchemyAsyncQueryService[source]¶
Bases:
ResultConverter
Simple service to execute the basic Query repository..
- __init__(session, **repo_kwargs)[source]¶
Configure the service object.
- Parameters:
session¶ (
Union
[AsyncSession
,async_scoped_session
[AsyncSession
]]) – Session managing the unit-of-work for the operation.**repo_kwargs¶ (
Any
) – Optional configuration values to pass into the repositorysession (AsyncSession | async_scoped_session[AsyncSession])
repo_kwargs (Any)
- Return type:
None
- classmethod new(cls, session=None, config=None)[source]¶
Context manager that returns instance of service object.
Handles construction of the database session._create_select_for_model
- Raises:
AdvancedAlchemyError – If no configuration or session is provided.
- Yields:
The service object instance.
- Return type:
AsyncIterator
[Self
]- Parameters:
session (AsyncSession | async_scoped_session[AsyncSession] | None)
config (SQLAlchemyAsyncConfig | None)
- class advanced_alchemy.service.SQLAlchemyAsyncRepositoryReadService[source]¶
Bases:
ResultConverter
,Generic
[ModelT
,SQLAlchemyAsyncRepositoryT
]Service object that operates on a repository object.
- __init__(session, *, statement=None, auto_expunge=False, auto_refresh=True, auto_commit=False, order_by=None, error_messages=Empty, wrap_exceptions=True, load=None, execution_options=None, uniquify=None, count_with_window_function=None, **repo_kwargs)[source]¶
Configure the service object.
- Parameters:
session¶ (
Union
[AsyncSession
,async_scoped_session
[AsyncSession
]]) – Session managing the unit-of-work for the operation.statement¶ (
Optional
[Select
]) – To facilitate customization of the underlying select query.auto_expunge¶ (
bool
) – Remove object from session before returning.auto_refresh¶ (
bool
) – Refresh object from session before returning.order_by¶ (
Union
[list
[Union
[tuple
[Union
[str
,InstrumentedAttribute
[Any
]],bool
],UnaryExpression
[Any
]]],tuple
[Union
[str
,InstrumentedAttribute
[Any
]],bool
],UnaryExpression
[Any
],None
]) – Set default order options for queries.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – A set of custom error messages to use for operationswrap_exceptions¶ (
bool
) – Wrap exceptions in a RepositoryErrorload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set default relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.count_with_window_function¶ (
Optional
[bool
]) – When false, list and count will use two queries instead of an analytical window function.**repo_kwargs¶ (
Any
) – passed as keyword args to repo instantiation.session (AsyncSession | async_scoped_session[AsyncSession])
statement (Select | None)
auto_expunge (bool)
auto_refresh (bool)
auto_commit (bool)
order_by (list[tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any]] | tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any] | None)
error_messages (ErrorMessages | type[Empty] | None)
wrap_exceptions (bool)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
count_with_window_function (bool | None)
repo_kwargs (Any)
- Return type:
None
- async count(*filters, statement=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, **kwargs)[source]¶
Count of records returned by query.
- Parameters:
*filters¶ (
Union
[StatementFilter
,ColumnElement
[bool
]]) – arguments for filtering.statement¶ (
Optional
[Select
]) – To facilitate customization of the underlying select query.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.filters (StatementFilter | ColumnElement[bool])
statement (Select | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
kwargs (Any)
- Return type:
- Returns:
A count of the collection, filtered, but ignoring pagination.
-
count_with_window_function:
ClassVar
[bool
] = True¶ Use an analytical window function to count results. This allows the count to be performed in a single query.
-
execution_options:
ClassVar
[Optional
[dict
[str
,Any
]]] = None¶ Default execution options for the repository.
- async exists(*filters, error_messages=Empty, load=None, execution_options=None, uniquify=None, **kwargs)[source]¶
Wrap repository exists operation.
- Parameters:
*filters¶ (
Union
[StatementFilter
,ColumnElement
[bool
]]) – Types for specific filtering operations.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.**kwargs¶ (
Any
) – Keyword arguments for attribute based filtering.filters (StatementFilter | ColumnElement[bool])
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
kwargs (Any)
- Return type:
- Returns:
Representation of instance with identifier item_id.
- async get(item_id, *, statement=None, id_attribute=None, auto_expunge=None, error_messages=Empty, load=None, execution_options=None, uniquify=None)[source]¶
Wrap repository scalar operation.
- Parameters:
auto_expunge¶ (
Optional
[bool
]) – Remove object from session before returning.statement¶ (
Optional
[Select
]) – To facilitate customization of the underlying select query.id_attribute¶ (
Union
[str
,InstrumentedAttribute
[Any
],None
]) – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.item_id (Any)
statement (Select | None)
id_attribute (str | InstrumentedAttribute[Any] | None)
auto_expunge (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
- Return type:
TypeVar
(ModelT
, bound= base.ModelProtocol)- Returns:
Representation of instance with identifier item_id.
- async get_one(*filters, statement=None, auto_expunge=None, load=None, error_messages=Empty, execution_options=None, uniquify=None, **kwargs)[source]¶
Wrap repository scalar operation.
- Parameters:
*filters¶ (
Union
[StatementFilter
,ColumnElement
[bool
]]) – Types for specific filtering operations.auto_expunge¶ (
Optional
[bool
]) – Remove object from session before returning.statement¶ (
Optional
[Select
]) – To facilitate customization of the underlying select query.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set default relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.**kwargs¶ (
Any
) – Identifier of the instance to be retrieved.filters (StatementFilter | ColumnElement[bool])
statement (Select | None)
auto_expunge (bool | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
error_messages (ErrorMessages | type[Empty] | None)
uniquify (bool | None)
kwargs (Any)
- Return type:
TypeVar
(ModelT
, bound= base.ModelProtocol)- Returns:
Representation of instance with identifier item_id.
- async get_one_or_none(*filters, statement=None, auto_expunge=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, **kwargs)[source]¶
Wrap repository scalar operation.
- Parameters:
*filters¶ (
Union
[StatementFilter
,ColumnElement
[bool
]]) – Types for specific filtering operations.auto_expunge¶ (
Optional
[bool
]) – Remove object from session before returning.statement¶ (
Optional
[Select
]) – To facilitate customization of the underlying select query.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set default relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.**kwargs¶ (
Any
) – Identifier of the instance to be retrieved.filters (StatementFilter | ColumnElement[bool])
statement (Select | None)
auto_expunge (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
kwargs (Any)
- Return type:
- Returns:
Representation of instance with identifier item_id.
- async list(*filters, statement=None, auto_expunge=None, order_by=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, **kwargs)[source]¶
Wrap repository scalars operation.
- Parameters:
*filters¶ (
Union
[StatementFilter
,ColumnElement
[bool
]]) – Types for specific filtering operations.auto_expunge¶ (
Optional
[bool
]) – Remove object from session before returning.statement¶ (
Optional
[Select
]) – To facilitate customization of the underlying select query.order_by¶ (
Union
[list
[Union
[tuple
[Union
[str
,InstrumentedAttribute
[Any
]],bool
],UnaryExpression
[Any
]]],tuple
[Union
[str
,InstrumentedAttribute
[Any
]],bool
],UnaryExpression
[Any
],None
]) – Set default order options for queries.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set default relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.filters (StatementFilter | ColumnElement[bool])
statement (Select | None)
auto_expunge (bool | None)
order_by (list[tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any]] | tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any] | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
kwargs (Any)
- Return type:
- Returns:
The list of instances retrieved from the repository.
- async list_and_count(*filters, statement=None, auto_expunge=None, count_with_window_function=None, order_by=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, **kwargs)[source]¶
List of records and total count returned by query.
- Parameters:
*filters¶ (
Union
[StatementFilter
,ColumnElement
[bool
]]) – Types for specific filtering operations.statement¶ (
Optional
[Select
]) – To facilitate customization of the underlying select query.auto_expunge¶ (
Optional
[bool
]) – Remove object from session before returning.count_with_window_function¶ (
Optional
[bool
]) – When false, list and count will use two queries instead of an analytical window function.order_by¶ (
Union
[list
[Union
[tuple
[Union
[str
,InstrumentedAttribute
[Any
]],bool
],UnaryExpression
[Any
]]],tuple
[Union
[str
,InstrumentedAttribute
[Any
]],bool
],UnaryExpression
[Any
],None
]) – Set default order options for queries.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.filters (StatementFilter | ColumnElement[bool])
statement (Select | None)
auto_expunge (bool | None)
count_with_window_function (bool | None)
order_by (list[tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any]] | tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any] | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
kwargs (Any)
- Return type:
tuple
[Sequence
[TypeVar
(ModelT
, bound= base.ModelProtocol)],int
]- Returns:
List of instances and count of total collection, ignoring pagination.
-
loader_options:
ClassVar
[Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]] = None¶ Default loader options for the repository.
-
match_fields:
ClassVar
[Union
[list
[str
],str
,None
]] = None¶ List of dialects that prefer to use
field.id = ANY(:1)
instead offield.id IN (...)
.
- classmethod new(cls, session=None, statement=None, config=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None, uniquify=None, count_with_window_function=None)[source]¶
Context manager that returns instance of service object.
Handles construction of the database session._create_select_for_model
- Raises:
AdvancedAlchemyError – If no configuration or session is provided.
- Yields:
The service object instance.
- Return type:
AsyncIterator
[Self
]- Parameters:
session (AsyncSession | async_scoped_session[AsyncSession] | None)
statement (Select | None)
config (SQLAlchemyAsyncConfig | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
count_with_window_function (bool | None)
- property repository: SQLAlchemyAsyncRepositoryT¶
Return the repository instance.
- Raises:
ImproperConfigurationError – If the repository is not initialized.
- Returns:
The repository instance.
- async to_model_on_create(data)[source]¶
Convenience method to allow for custom behavior on create.
- Parameters:
data¶ – The data to be converted to a model.
data (ModelDictT[ModelT])
- Returns:
The data to be converted to a model.
- Return type:
ModelDictT[ModelT]
- async to_model_on_delete(data)[source]¶
Convenience method to allow for custom behavior on delete.
- Parameters:
data¶ – The data to be converted to a model.
data (ModelDictT[ModelT])
- Returns:
The data to be converted to a model.
- Return type:
ModelDictT[ModelT]
- async to_model_on_update(data)[source]¶
Convenience method to allow for custom behavior on update.
- Parameters:
data¶ – The data to be converted to a model.
data (ModelDictT[ModelT])
- Returns:
The data to be converted to a model.
- Return type:
ModelDictT[ModelT]
- class advanced_alchemy.service.SQLAlchemyAsyncRepositoryService[source]¶
Bases:
SQLAlchemyAsyncRepositoryReadService
[ModelT
,SQLAlchemyAsyncRepositoryT
],Generic
[ModelT
,SQLAlchemyAsyncRepositoryT
]Service object that operates on a repository object.
- async create(data, *, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=Empty)[source]¶
Wrap repository instance creation.
- Parameters:
data¶ – Representation to be created.
auto_expunge¶ – Remove object from session before returning.
auto_refresh¶ – Refresh object from session before returning.
auto_commit¶ – Commit objects before returning.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
data (ModelDictT[ModelT])
auto_commit (bool | None)
auto_expunge (bool | None)
auto_refresh (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
- Returns:
Representation of created instance.
- Return type:
ModelT
- async create_many(data, *, auto_commit=None, auto_expunge=None, error_messages=Empty)[source]¶
Wrap repository bulk instance creation.
- Parameters:
data¶ – Representations to be created.
auto_expunge¶ – Remove object from session before returning.
auto_commit¶ – Commit objects before returning.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
data (BulkModelDictT[ModelT])
auto_commit (bool | None)
auto_expunge (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
- Returns:
Representation of created instances.
- Return type:
Sequence[ModelT]
- async delete(item_id, *, auto_commit=None, auto_expunge=None, id_attribute=None, error_messages=Empty, load=None, execution_options=None, uniquify=None)[source]¶
Wrap repository delete operation.
- Parameters:
auto_commit¶ (
Optional
[bool
]) – Commit objects before returning.auto_expunge¶ (
Optional
[bool
]) – Remove object from session before returning.id_attribute¶ (
Union
[str
,InstrumentedAttribute
[Any
],None
]) – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set default relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.item_id (Any)
auto_commit (bool | None)
auto_expunge (bool | None)
id_attribute (str | InstrumentedAttribute[Any] | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
- Return type:
TypeVar
(ModelT
, bound= base.ModelProtocol)- Returns:
Representation of the deleted instance.
- async delete_many(item_ids, *, auto_commit=None, auto_expunge=None, id_attribute=None, chunk_size=None, error_messages=Empty, load=None, execution_options=None, uniquify=None)[source]¶
Wrap repository bulk instance deletion.
- Parameters:
item_ids¶ (
list
[Any
]) – Identifier of instance to be deleted.auto_expunge¶ (
Optional
[bool
]) – Remove object from session before returning.auto_commit¶ (
Optional
[bool
]) – Commit objects before returning.id_attribute¶ (
Union
[str
,InstrumentedAttribute
[Any
],None
]) – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.chunk_size¶ (
Optional
[int
]) – Allows customization of theinsertmanyvalues_max_parameters
setting for the driver. Defaults to 950 if left unset.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set default relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.auto_commit (bool | None)
auto_expunge (bool | None)
id_attribute (str | InstrumentedAttribute[Any] | None)
chunk_size (int | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
- Return type:
- Returns:
Representation of removed instances.
- async delete_where(*filters, auto_commit=None, auto_expunge=None, error_messages=Empty, sanity_check=True, load=None, execution_options=None, uniquify=None, **kwargs)[source]¶
Wrap repository scalars operation.
- Parameters:
*filters¶ (
Union
[StatementFilter
,ColumnElement
[bool
]]) – Types for specific filtering operations.auto_expunge¶ (
Optional
[bool
]) – Remove object from session before returning.auto_commit¶ (
Optional
[bool
]) – Commit objects before returning.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientssanity_check¶ (
bool
) – When true, the length of selected instances is compared to the deleted row countload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set default relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.filters (StatementFilter | ColumnElement[bool])
auto_commit (bool | None)
auto_expunge (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
sanity_check (bool)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
kwargs (Any)
- Return type:
- Returns:
The list of instances deleted from the repository.
- async get_and_update(*filters, match_fields=None, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, **kwargs)[source]¶
Wrap repository instance creation.
- Parameters:
*filters¶ (
Union
[StatementFilter
,ColumnElement
[bool
]]) – Types for specific filtering operations.match_fields¶ (
Union
[list
[str
],str
,None
]) – a list of keys to use to match the existing model. When empty, all fields are matched.attribute_names¶ (
Optional
[Iterable
[str
]]) – an iterable of attribute names to pass into theupdate
method.with_for_update¶ (
Union
[sqlalchemy.sql.selectable.ForUpdateArg,None
,bool
,Dict
[str
,Any
]]) – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECTauto_expunge¶ (
Optional
[bool
]) – Remove object from session before returning.auto_refresh¶ (
Optional
[bool
]) – Refresh object from session before returning.auto_commit¶ (
Optional
[bool
]) – Commit objects before returning.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set default relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.**kwargs¶ (
Any
) – Identifier of the instance to be retrieved.filters (StatementFilter | ColumnElement[bool])
with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])
auto_commit (bool | None)
auto_expunge (bool | None)
auto_refresh (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
kwargs (Any)
- Return type:
- Returns:
Representation of updated instance.
- async get_or_upsert(*filters, match_fields=None, upsert=True, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, **kwargs)[source]¶
Wrap repository instance creation.
- Parameters:
*filters¶ (
Union
[StatementFilter
,ColumnElement
[bool
]]) – Types for specific filtering operations.match_fields¶ (
Union
[list
[str
],str
,None
]) – a list of keys to use to match the existing model. When empty, all fields are matched.upsert¶ (
bool
) – When using match_fields and actual model values differ from kwargs, perform an update operation on the model.create¶ – Should a model be created. If no model is found, an exception is raised.
attribute_names¶ (
Optional
[Iterable
[str
]]) – an iterable of attribute names to pass into theupdate
method.with_for_update¶ (
Union
[sqlalchemy.sql.selectable.ForUpdateArg,None
,bool
,Dict
[str
,Any
]]) – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECTauto_expunge¶ (
Optional
[bool
]) – Remove object from session before returning.auto_refresh¶ (
Optional
[bool
]) – Refresh object from session before returning.auto_commit¶ (
Optional
[bool
]) – Commit objects before returning.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set default relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.**kwargs¶ (
Any
) – Identifier of the instance to be retrieved.filters (StatementFilter | ColumnElement[bool])
upsert (bool)
with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])
auto_commit (bool | None)
auto_expunge (bool | None)
auto_refresh (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
kwargs (Any)
- Return type:
- Returns:
Representation of created instance.
- async update(data, item_id=None, *, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, id_attribute=None, error_messages=Empty, load=None, execution_options=None, uniquify=None)[source]¶
Wrap repository update operation.
- Parameters:
data¶ – Representation to be updated.
item_id¶ – Identifier of item to be updated.
attribute_names¶ – an iterable of attribute names to pass into the
update
method.with_for_update¶ – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT
auto_expunge¶ – Remove object from session before returning.
auto_refresh¶ – Refresh object from session before returning.
auto_commit¶ – Commit objects before returning.
id_attribute¶ – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
uniquify¶ – Optionally apply the
unique()
method to results before returning.data (ModelDictT[ModelT])
item_id (Any | None)
with_for_update (ForUpdateArg | None | bool | Dict[str, Any])
auto_commit (bool | None)
auto_expunge (bool | None)
auto_refresh (bool | None)
id_attribute (str | InstrumentedAttribute[Any] | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
- Raises:
RepositoryError – If no configuration or session is provided.
- Returns:
Updated representation.
- Return type:
ModelT
- async update_many(data, *, auto_commit=None, auto_expunge=None, error_messages=Empty, load=None, execution_options=None, uniquify=None)[source]¶
Wrap repository bulk instance update.
- Parameters:
data¶ – Representations to be updated.
auto_expunge¶ – Remove object from session before returning.
auto_commit¶ – Commit objects before returning.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
uniquify¶ – Optionally apply the
unique()
method to results before returning.data (BulkModelDictT[ModelT])
auto_commit (bool | None)
auto_expunge (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
- Returns:
Representation of updated instances.
- Return type:
Sequence[ModelT]
- async upsert(data, item_id=None, *, attribute_names=None, with_for_update=None, auto_expunge=None, auto_commit=None, auto_refresh=None, match_fields=None, error_messages=Empty, load=None, execution_options=None, uniquify=None)[source]¶
Wrap repository upsert operation.
- Parameters:
data¶ – Instance to update existing, or be created. Identifier used to determine if an existing instance exists is the value of an attribute on data named as value of self.id_attribute.
item_id¶ – Identifier of the object for upsert.
attribute_names¶ – an iterable of attribute names to pass into the
update
method.with_for_update¶ – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT
auto_expunge¶ – Remove object from session before returning.
auto_refresh¶ – Refresh object from session before returning.
auto_commit¶ – Commit objects before returning.
match_fields¶ – a list of keys to use to match the existing model. When empty, all fields are matched.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
uniquify¶ – Optionally apply the
unique()
method to results before returning.data (ModelDictT[ModelT])
item_id (Any | None)
with_for_update (ForUpdateArg | None | bool | Dict[str, Any])
auto_expunge (bool | None)
auto_commit (bool | None)
auto_refresh (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
- Returns:
Updated or created representation.
- Return type:
ModelT
- async upsert_many(data, *, auto_expunge=None, auto_commit=None, no_merge=False, match_fields=None, error_messages=Empty, load=None, execution_options=None, uniquify=None)[source]¶
Wrap repository upsert operation.
- Parameters:
data¶ – Instance to update existing, or be created.
auto_expunge¶ – Remove object from session before returning.
auto_commit¶ – Commit objects before returning.
no_merge¶ – Skip the usage of optimized Merge statements (reserved for future use)
match_fields¶ – a list of keys to use to match the existing model. When empty, all fields are matched.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
uniquify¶ – Optionally apply the
unique()
method to results before returning.data (BulkModelDictT[ModelT])
auto_expunge (bool | None)
auto_commit (bool | None)
no_merge (bool)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
- Returns:
Updated or created representation.
- Return type:
Sequence[ModelT]
- class advanced_alchemy.service.SQLAlchemySyncQueryService[source]¶
Bases:
ResultConverter
Simple service to execute the basic Query repository..
- __init__(session, **repo_kwargs)[source]¶
Configure the service object.
- Parameters:
session¶ (
Union
[Session
,scoped_session
[Session
]]) – Session managing the unit-of-work for the operation.**repo_kwargs¶ (
Any
) – Optional configuration values to pass into the repositorysession (Session | scoped_session[Session])
repo_kwargs (Any)
- Return type:
None
- classmethod new(cls, session=None, config=None)[source]¶
Context manager that returns instance of service object.
Handles construction of the database session._create_select_for_model
- Raises:
AdvancedAlchemyError – If no configuration or session is provided.
- Yields:
The service object instance.
- Return type:
Iterator
[Self
]- Parameters:
session (Session | scoped_session[Session] | None)
config (SQLAlchemySyncConfig | None)
- class advanced_alchemy.service.SQLAlchemySyncRepositoryReadService[source]¶
Bases:
ResultConverter
,Generic
[ModelT
,SQLAlchemySyncRepositoryT
]Service object that operates on a repository object.
- __init__(session, *, statement=None, auto_expunge=False, auto_refresh=True, auto_commit=False, order_by=None, error_messages=Empty, wrap_exceptions=True, load=None, execution_options=None, uniquify=None, count_with_window_function=None, **repo_kwargs)[source]¶
Configure the service object.
- Parameters:
session¶ (
Union
[Session
,scoped_session
[Session
]]) – Session managing the unit-of-work for the operation.statement¶ (
Optional
[Select
]) – To facilitate customization of the underlying select query.auto_expunge¶ (
bool
) – Remove object from session before returning.auto_refresh¶ (
bool
) – Refresh object from session before returning.order_by¶ (
Union
[list
[Union
[tuple
[Union
[str
,InstrumentedAttribute
[Any
]],bool
],UnaryExpression
[Any
]]],tuple
[Union
[str
,InstrumentedAttribute
[Any
]],bool
],UnaryExpression
[Any
],None
]) – Set default order options for queries.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – A set of custom error messages to use for operationswrap_exceptions¶ (
bool
) – Wrap exceptions in a RepositoryErrorload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set default relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.count_with_window_function¶ (
Optional
[bool
]) – When false, list and count will use two queries instead of an analytical window function.**repo_kwargs¶ (
Any
) – passed as keyword args to repo instantiation.session (Session | scoped_session[Session])
statement (Select | None)
auto_expunge (bool)
auto_refresh (bool)
auto_commit (bool)
order_by (list[tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any]] | tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any] | None)
error_messages (ErrorMessages | type[Empty] | None)
wrap_exceptions (bool)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
count_with_window_function (bool | None)
repo_kwargs (Any)
- Return type:
None
- count(*filters, statement=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, **kwargs)[source]¶
Count of records returned by query.
- Parameters:
*filters¶ (
Union
[StatementFilter
,ColumnElement
[bool
]]) – arguments for filtering.statement¶ (
Optional
[Select
]) – To facilitate customization of the underlying select query.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.filters (StatementFilter | ColumnElement[bool])
statement (Select | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
kwargs (Any)
- Return type:
- Returns:
A count of the collection, filtered, but ignoring pagination.
-
count_with_window_function:
ClassVar
[bool
] = True¶ Use an analytical window function to count results. This allows the count to be performed in a single query.
-
execution_options:
ClassVar
[Optional
[dict
[str
,Any
]]] = None¶ Default execution options for the repository.
- exists(*filters, error_messages=Empty, load=None, execution_options=None, uniquify=None, **kwargs)[source]¶
Wrap repository exists operation.
- Parameters:
*filters¶ (
Union
[StatementFilter
,ColumnElement
[bool
]]) – Types for specific filtering operations.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.**kwargs¶ (
Any
) – Keyword arguments for attribute based filtering.filters (StatementFilter | ColumnElement[bool])
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
kwargs (Any)
- Return type:
- Returns:
Representation of instance with identifier item_id.
- get(item_id, *, statement=None, id_attribute=None, auto_expunge=None, error_messages=Empty, load=None, execution_options=None, uniquify=None)[source]¶
Wrap repository scalar operation.
- Parameters:
auto_expunge¶ (
Optional
[bool
]) – Remove object from session before returning.statement¶ (
Optional
[Select
]) – To facilitate customization of the underlying select query.id_attribute¶ (
Union
[str
,InstrumentedAttribute
[Any
],None
]) – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.item_id (Any)
statement (Select | None)
id_attribute (str | InstrumentedAttribute[Any] | None)
auto_expunge (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
- Return type:
TypeVar
(ModelT
, bound= base.ModelProtocol)- Returns:
Representation of instance with identifier item_id.
- get_one(*filters, statement=None, auto_expunge=None, load=None, error_messages=Empty, execution_options=None, uniquify=None, **kwargs)[source]¶
Wrap repository scalar operation.
- Parameters:
*filters¶ (
Union
[StatementFilter
,ColumnElement
[bool
]]) – Types for specific filtering operations.auto_expunge¶ (
Optional
[bool
]) – Remove object from session before returning.statement¶ (
Optional
[Select
]) – To facilitate customization of the underlying select query.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set default relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.**kwargs¶ (
Any
) – Identifier of the instance to be retrieved.filters (StatementFilter | ColumnElement[bool])
statement (Select | None)
auto_expunge (bool | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
error_messages (ErrorMessages | type[Empty] | None)
uniquify (bool | None)
kwargs (Any)
- Return type:
TypeVar
(ModelT
, bound= base.ModelProtocol)- Returns:
Representation of instance with identifier item_id.
- get_one_or_none(*filters, statement=None, auto_expunge=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, **kwargs)[source]¶
Wrap repository scalar operation.
- Parameters:
*filters¶ (
Union
[StatementFilter
,ColumnElement
[bool
]]) – Types for specific filtering operations.auto_expunge¶ (
Optional
[bool
]) – Remove object from session before returning.statement¶ (
Optional
[Select
]) – To facilitate customization of the underlying select query.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set default relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.**kwargs¶ (
Any
) – Identifier of the instance to be retrieved.filters (StatementFilter | ColumnElement[bool])
statement (Select | None)
auto_expunge (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
kwargs (Any)
- Return type:
- Returns:
Representation of instance with identifier item_id.
- list(*filters, statement=None, auto_expunge=None, order_by=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, **kwargs)[source]¶
Wrap repository scalars operation.
- Parameters:
*filters¶ (
Union
[StatementFilter
,ColumnElement
[bool
]]) – Types for specific filtering operations.auto_expunge¶ (
Optional
[bool
]) – Remove object from session before returning.statement¶ (
Optional
[Select
]) – To facilitate customization of the underlying select query.order_by¶ (
Union
[list
[Union
[tuple
[Union
[str
,InstrumentedAttribute
[Any
]],bool
],UnaryExpression
[Any
]]],tuple
[Union
[str
,InstrumentedAttribute
[Any
]],bool
],UnaryExpression
[Any
],None
]) – Set default order options for queries.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set default relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.filters (StatementFilter | ColumnElement[bool])
statement (Select | None)
auto_expunge (bool | None)
order_by (list[tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any]] | tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any] | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
kwargs (Any)
- Return type:
- Returns:
The list of instances retrieved from the repository.
- list_and_count(*filters, statement=None, auto_expunge=None, count_with_window_function=None, order_by=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, **kwargs)[source]¶
List of records and total count returned by query.
- Parameters:
*filters¶ (
Union
[StatementFilter
,ColumnElement
[bool
]]) – Types for specific filtering operations.statement¶ (
Optional
[Select
]) – To facilitate customization of the underlying select query.auto_expunge¶ (
Optional
[bool
]) – Remove object from session before returning.count_with_window_function¶ (
Optional
[bool
]) – When false, list and count will use two queries instead of an analytical window function.order_by¶ (
Union
[list
[Union
[tuple
[Union
[str
,InstrumentedAttribute
[Any
]],bool
],UnaryExpression
[Any
]]],tuple
[Union
[str
,InstrumentedAttribute
[Any
]],bool
],UnaryExpression
[Any
],None
]) – Set default order options for queries.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.filters (StatementFilter | ColumnElement[bool])
statement (Select | None)
auto_expunge (bool | None)
count_with_window_function (bool | None)
order_by (list[tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any]] | tuple[str | InstrumentedAttribute[Any], bool] | UnaryExpression[Any] | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
kwargs (Any)
- Return type:
tuple
[Sequence
[TypeVar
(ModelT
, bound= base.ModelProtocol)],int
]- Returns:
List of instances and count of total collection, ignoring pagination.
-
loader_options:
ClassVar
[Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]] = None¶ Default loader options for the repository.
-
match_fields:
ClassVar
[Union
[list
[str
],str
,None
]] = None¶ List of dialects that prefer to use
field.id = ANY(:1)
instead offield.id IN (...)
.
- classmethod new(cls, session=None, statement=None, config=None, error_messages=<class 'advanced_alchemy.utils.dataclass.Empty'>, load=None, execution_options=None, uniquify=None, count_with_window_function=None)[source]¶
Context manager that returns instance of service object.
Handles construction of the database session._create_select_for_model
- Raises:
AdvancedAlchemyError – If no configuration or session is provided.
- Yields:
The service object instance.
- Return type:
Iterator
[Self
]- Parameters:
session (Session | scoped_session[Session] | None)
statement (Select | None)
config (SQLAlchemySyncConfig | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
count_with_window_function (bool | None)
- property repository: SQLAlchemySyncRepositoryT¶
Return the repository instance.
- Raises:
ImproperConfigurationError – If the repository is not initialized.
- Returns:
The repository instance.
- to_model_on_create(data)[source]¶
Convenience method to allow for custom behavior on create.
- Parameters:
data¶ – The data to be converted to a model.
data (ModelDictT[ModelT])
- Returns:
The data to be converted to a model.
- Return type:
ModelDictT[ModelT]
- to_model_on_delete(data)[source]¶
Convenience method to allow for custom behavior on delete.
- Parameters:
data¶ – The data to be converted to a model.
data (ModelDictT[ModelT])
- Returns:
The data to be converted to a model.
- Return type:
ModelDictT[ModelT]
- to_model_on_update(data)[source]¶
Convenience method to allow for custom behavior on update.
- Parameters:
data¶ – The data to be converted to a model.
data (ModelDictT[ModelT])
- Returns:
The data to be converted to a model.
- Return type:
ModelDictT[ModelT]
- class advanced_alchemy.service.SQLAlchemySyncRepositoryService[source]¶
Bases:
SQLAlchemySyncRepositoryReadService
[ModelT
,SQLAlchemySyncRepositoryT
],Generic
[ModelT
,SQLAlchemySyncRepositoryT
]Service object that operates on a repository object.
- create(data, *, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=Empty)[source]¶
Wrap repository instance creation.
- Parameters:
data¶ – Representation to be created.
auto_expunge¶ – Remove object from session before returning.
auto_refresh¶ – Refresh object from session before returning.
auto_commit¶ – Commit objects before returning.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
data (ModelDictT[ModelT])
auto_commit (bool | None)
auto_expunge (bool | None)
auto_refresh (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
- Returns:
Representation of created instance.
- Return type:
ModelT
- create_many(data, *, auto_commit=None, auto_expunge=None, error_messages=Empty)[source]¶
Wrap repository bulk instance creation.
- Parameters:
data¶ – Representations to be created.
auto_expunge¶ – Remove object from session before returning.
auto_commit¶ – Commit objects before returning.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
data (BulkModelDictT[ModelT])
auto_commit (bool | None)
auto_expunge (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
- Returns:
Representation of created instances.
- Return type:
Sequence[ModelT]
- delete(item_id, *, auto_commit=None, auto_expunge=None, id_attribute=None, error_messages=Empty, load=None, execution_options=None, uniquify=None)[source]¶
Wrap repository delete operation.
- Parameters:
auto_commit¶ (
Optional
[bool
]) – Commit objects before returning.auto_expunge¶ (
Optional
[bool
]) – Remove object from session before returning.id_attribute¶ (
Union
[str
,InstrumentedAttribute
[Any
],None
]) – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set default relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.item_id (Any)
auto_commit (bool | None)
auto_expunge (bool | None)
id_attribute (str | InstrumentedAttribute[Any] | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
- Return type:
TypeVar
(ModelT
, bound= base.ModelProtocol)- Returns:
Representation of the deleted instance.
- delete_many(item_ids, *, auto_commit=None, auto_expunge=None, id_attribute=None, chunk_size=None, error_messages=Empty, load=None, execution_options=None, uniquify=None)[source]¶
Wrap repository bulk instance deletion.
- Parameters:
item_ids¶ (
list
[Any
]) – Identifier of instance to be deleted.auto_expunge¶ (
Optional
[bool
]) – Remove object from session before returning.auto_commit¶ (
Optional
[bool
]) – Commit objects before returning.id_attribute¶ (
Union
[str
,InstrumentedAttribute
[Any
],None
]) – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.chunk_size¶ (
Optional
[int
]) – Allows customization of theinsertmanyvalues_max_parameters
setting for the driver. Defaults to 950 if left unset.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set default relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.auto_commit (bool | None)
auto_expunge (bool | None)
id_attribute (str | InstrumentedAttribute[Any] | None)
chunk_size (int | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
- Return type:
- Returns:
Representation of removed instances.
- delete_where(*filters, auto_commit=None, auto_expunge=None, error_messages=Empty, sanity_check=True, load=None, execution_options=None, uniquify=None, **kwargs)[source]¶
Wrap repository scalars operation.
- Parameters:
*filters¶ (
Union
[StatementFilter
,ColumnElement
[bool
]]) – Types for specific filtering operations.auto_expunge¶ (
Optional
[bool
]) – Remove object from session before returning.auto_commit¶ (
Optional
[bool
]) – Commit objects before returning.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientssanity_check¶ (
bool
) – When true, the length of selected instances is compared to the deleted row countload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set default relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.filters (StatementFilter | ColumnElement[bool])
auto_commit (bool | None)
auto_expunge (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
sanity_check (bool)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
kwargs (Any)
- Return type:
- Returns:
The list of instances deleted from the repository.
- get_and_update(*filters, match_fields=None, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, **kwargs)[source]¶
Wrap repository instance creation.
- Parameters:
*filters¶ (
Union
[StatementFilter
,ColumnElement
[bool
]]) – Types for specific filtering operations.match_fields¶ (
Union
[list
[str
],str
,None
]) – a list of keys to use to match the existing model. When empty, all fields are matched.attribute_names¶ (
Optional
[Iterable
[str
]]) – an iterable of attribute names to pass into theupdate
method.with_for_update¶ (
Union
[sqlalchemy.sql.selectable.ForUpdateArg,None
,bool
,Dict
[str
,Any
]]) – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECTauto_expunge¶ (
Optional
[bool
]) – Remove object from session before returning.auto_refresh¶ (
Optional
[bool
]) – Refresh object from session before returning.auto_commit¶ (
Optional
[bool
]) – Commit objects before returning.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set default relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.**kwargs¶ (
Any
) – Identifier of the instance to be retrieved.filters (StatementFilter | ColumnElement[bool])
with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])
auto_commit (bool | None)
auto_expunge (bool | None)
auto_refresh (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
kwargs (Any)
- Return type:
- Returns:
Representation of updated instance.
- get_or_upsert(*filters, match_fields=None, upsert=True, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, error_messages=Empty, load=None, execution_options=None, uniquify=None, **kwargs)[source]¶
Wrap repository instance creation.
- Parameters:
*filters¶ (
Union
[StatementFilter
,ColumnElement
[bool
]]) – Types for specific filtering operations.match_fields¶ (
Union
[list
[str
],str
,None
]) – a list of keys to use to match the existing model. When empty, all fields are matched.upsert¶ (
bool
) – When using match_fields and actual model values differ from kwargs, perform an update operation on the model.create¶ – Should a model be created. If no model is found, an exception is raised.
attribute_names¶ (
Optional
[Iterable
[str
]]) – an iterable of attribute names to pass into theupdate
method.with_for_update¶ (
Union
[sqlalchemy.sql.selectable.ForUpdateArg,None
,bool
,Dict
[str
,Any
]]) – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECTauto_expunge¶ (
Optional
[bool
]) – Remove object from session before returning.auto_refresh¶ (
Optional
[bool
]) – Refresh object from session before returning.auto_commit¶ (
Optional
[bool
]) – Commit objects before returning.error_messages¶ (
Union
[ErrorMessages
,type
[Empty
],None
]) – An optional dictionary of templates to use for friendlier error messages to clientsload¶ (
Union
[Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],Sequence
[Union
[_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
]]]]],_AbstractLoad
,Literal
['*'
],InstrumentedAttribute
[Any
],RelationshipProperty
[Any
],MapperProperty
[Any
],ExecutableOption
,Sequence
[ExecutableOption
],None
]) – Set default relationships to be loadedexecution_options¶ (
Optional
[dict
[str
,Any
]]) – Set default execution optionsuniquify¶ (
Optional
[bool
]) – Optionally apply theunique()
method to results before returning.**kwargs¶ (
Any
) – Identifier of the instance to be retrieved.filters (StatementFilter | ColumnElement[bool])
upsert (bool)
with_for_update (TypeAliasForwardRef('sqlalchemy.sql.selectable.ForUpdateArg') | None | bool | Dict[str, Any])
auto_commit (bool | None)
auto_expunge (bool | None)
auto_refresh (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
kwargs (Any)
- Return type:
- Returns:
Representation of created instance.
- update(data, item_id=None, *, attribute_names=None, with_for_update=None, auto_commit=None, auto_expunge=None, auto_refresh=None, id_attribute=None, error_messages=Empty, load=None, execution_options=None, uniquify=None)[source]¶
Wrap repository update operation.
- Parameters:
data¶ – Representation to be updated.
item_id¶ – Identifier of item to be updated.
attribute_names¶ – an iterable of attribute names to pass into the
update
method.with_for_update¶ – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT
auto_expunge¶ – Remove object from session before returning.
auto_refresh¶ – Refresh object from session before returning.
auto_commit¶ – Commit objects before returning.
id_attribute¶ – Allows customization of the unique identifier to use for model fetching. Defaults to id, but can reference any surrogate or candidate key for the table.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
uniquify¶ – Optionally apply the
unique()
method to results before returning.data (ModelDictT[ModelT])
item_id (Any | None)
with_for_update (ForUpdateArg | None | bool | Dict[str, Any])
auto_commit (bool | None)
auto_expunge (bool | None)
auto_refresh (bool | None)
id_attribute (str | InstrumentedAttribute[Any] | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
- Raises:
RepositoryError – If no configuration or session is provided.
- Returns:
Updated representation.
- Return type:
ModelT
- update_many(data, *, auto_commit=None, auto_expunge=None, error_messages=Empty, load=None, execution_options=None, uniquify=None)[source]¶
Wrap repository bulk instance update.
- Parameters:
data¶ – Representations to be updated.
auto_expunge¶ – Remove object from session before returning.
auto_commit¶ – Commit objects before returning.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
uniquify¶ – Optionally apply the
unique()
method to results before returning.data (BulkModelDictT[ModelT])
auto_commit (bool | None)
auto_expunge (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
- Returns:
Representation of updated instances.
- Return type:
Sequence[ModelT]
- upsert(data, item_id=None, *, attribute_names=None, with_for_update=None, auto_expunge=None, auto_commit=None, auto_refresh=None, match_fields=None, error_messages=Empty, load=None, execution_options=None, uniquify=None)[source]¶
Wrap repository upsert operation.
- Parameters:
data¶ – Instance to update existing, or be created. Identifier used to determine if an existing instance exists is the value of an attribute on data named as value of self.id_attribute.
item_id¶ – Identifier of the object for upsert.
attribute_names¶ – an iterable of attribute names to pass into the
update
method.with_for_update¶ – indicating FOR UPDATE should be used, or may be a dictionary containing flags to indicate a more specific set of FOR UPDATE flags for the SELECT
auto_expunge¶ – Remove object from session before returning.
auto_refresh¶ – Refresh object from session before returning.
auto_commit¶ – Commit objects before returning.
match_fields¶ – a list of keys to use to match the existing model. When empty, all fields are matched.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
uniquify¶ – Optionally apply the
unique()
method to results before returning.data (ModelDictT[ModelT])
item_id (Any | None)
with_for_update (ForUpdateArg | None | bool | Dict[str, Any])
auto_expunge (bool | None)
auto_commit (bool | None)
auto_refresh (bool | None)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
- Returns:
Updated or created representation.
- Return type:
ModelT
- upsert_many(data, *, auto_expunge=None, auto_commit=None, no_merge=False, match_fields=None, error_messages=Empty, load=None, execution_options=None, uniquify=None)[source]¶
Wrap repository upsert operation.
- Parameters:
data¶ – Instance to update existing, or be created.
auto_expunge¶ – Remove object from session before returning.
auto_commit¶ – Commit objects before returning.
no_merge¶ – Skip the usage of optimized Merge statements (reserved for future use)
match_fields¶ – a list of keys to use to match the existing model. When empty, all fields are matched.
error_messages¶ – An optional dictionary of templates to use for friendlier error messages to clients
load¶ – Set default relationships to be loaded
execution_options¶ – Set default execution options
uniquify¶ – Optionally apply the
unique()
method to results before returning.data (BulkModelDictT[ModelT])
auto_expunge (bool | None)
auto_commit (bool | None)
no_merge (bool)
error_messages (ErrorMessages | type[Empty] | None)
load (Sequence[_AbstractLoad | Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None)
uniquify (bool | None)
- Returns:
Updated or created representation.
- Return type:
Sequence[ModelT]
- advanced_alchemy.service.fields(cls)[source]¶
Return the tuple of attrs attributes for a class.
The tuple also allows accessing the fields by their names (see below for examples).
- Parameters:
- Raises:
TypeError – If cls is not a class.
attrs.exceptions.NotAnAttrsClassError – If cls is not an attrs class.
- Returns:
tuple (with name accessors) of attrs.Attribute
Changed in version 16.2.0: Returned tuple allows accessing the fields by name.
Changed in version 23.1.0: Add support for generic classes.
- advanced_alchemy.service.find_filter(filter_type, filters)[source]¶
Get the filter specified by filter type from the filters.
- Parameters:
filter_type¶ – The type of filter to find.
filters¶ – filter types to apply to the query
filter_type (type[FilterTypeT])
filters (Union[Sequence[Union[StatementFilter, ColumnElement[bool]]], Sequence[StatementFilter]])
- Returns:
The match filter instance or None
- Return type:
Union[FilterTypeT, None]
- advanced_alchemy.service.is_attrs_instance(obj)[source]¶
Check if a value is an attrs class instance.
- advanced_alchemy.service.is_attrs_instance_with_field(v, field_name)[source]¶
Check if an attrs instance has a specific field.
- advanced_alchemy.service.is_attrs_instance_without_field(v, field_name)[source]¶
Check if an attrs instance does not have a specific field.
- advanced_alchemy.service.is_dict_with_field(v, field_name)[source]¶
Check if a dictionary has a specific field.
- advanced_alchemy.service.is_dict_without_field(v, field_name)[source]¶
Check if a dictionary does not have a specific field.
- advanced_alchemy.service.is_msgspec_struct_with_field(v, field_name)[source]¶
Check if a msgspec struct has a specific field.
- advanced_alchemy.service.is_msgspec_struct_without_field(v, field_name)[source]¶
Check if a msgspec struct does not have a specific field.
- advanced_alchemy.service.is_pydantic_model_with_field(v, field_name)[source]¶
Check if a pydantic model has a specific field.
- advanced_alchemy.service.is_pydantic_model_without_field(v, field_name)[source]¶
Check if a pydantic model does not have a specific field.
- advanced_alchemy.service.is_schema(v)[source]¶
Check if a value is a msgspec Struct, Pydantic model, or attrs instance.
- advanced_alchemy.service.is_schema_or_dict(v)[source]¶
Check if a value is a msgspec Struct, Pydantic model, attrs class, or dict.
- advanced_alchemy.service.is_schema_or_dict_with_field(v, field_name)[source]¶
Check if a value is a msgspec Struct, Pydantic model, attrs instance, or dict with a specific field.
- advanced_alchemy.service.is_schema_or_dict_without_field(v, field_name)[source]¶
Check if a value is a msgspec Struct, Pydantic model, attrs instance, or dict without a specific field.
- advanced_alchemy.service.is_schema_with_field(v, field_name)[source]¶
Check if a value is a msgspec Struct, Pydantic model, or attrs instance with a specific field.
- advanced_alchemy.service.is_schema_without_field(v, field_name)[source]¶
Check if a value is a msgspec Struct, Pydantic model, or attrs instance without a specific field.
- advanced_alchemy.service.model_from_dict(model, **kwargs)[source]¶
Create an ORM model instance from a dictionary of attributes.
- Parameters:
- Returns:
A new instance of the model populated with the provided values.
- Return type:
ModelT
- advanced_alchemy.service.schema_dump(data, exclude_unset=True)[source]¶
Dump a data object to a dictionary.
- Parameters:
data¶ –
dict[str, Any]
|advanced_alchemy.base.ModelProtocol
|msgspec.Struct
|pydantic.BaseModel
|attrs class
|litestar.dto.data_structures.DTOData[ModelT]
|sqlalchemy.RowMapping
|sqlalchemy.engine.row.Row
exclude_unset¶ –
bool
Whether to exclude unset values.data (Union[dict[str, Any], ModelT, SupportedSchemaModel, DTODataLike[ModelT], RowMapping, Row[Any]])
exclude_unset (bool)
- Returns:
type: dict[str, Any],
ModelProtocol
]- Return type:
Union[