py4ai.data.layer.sqlalchemy.repository module
Module for SQL Alchemy repository implementations.
- class py4ai.data.layer.sqlalchemy.repository.SqlAlchemyRepository(engine: AsyncEngine, serializer: SqlAlchemySerializer[KE, KD, E])
Bases:
Repository[KE,KD,E,Dict[str,Any],Union[BinaryExpression,BooleanClauseList]],Generic[KE,KD,E]Repository implementation for SQL Alchemy persistence layers.
Return a instance of the SQL Alchemy Repository.
- Parameters
engine – SQL Alchemy Async Engine to be used for db connections
serializer – Data serializer for serializing/deserializing raw data
- async create(entity: E) E
Create the entity in the underlying persistence layer.
- Parameters
entity – Entity to be created
- Returns
same entity provided as input, after creation. If creation fails, an error should be returned.
- Raises
ValueError – if the object is not inserted correctly
- async delete(key: KE) bool
Delete the entry in the persisence layer associated to the provided entity key.
- Parameters
key – key identifying the entity.
- Returns
boolean value indicating whether the deletion has completed successfully.
- async delete_by_criteria(criteria: SearchCriteria[Union[BinaryExpression, BooleanClauseList]]) bool
Delete all entries matching a given query.
- Parameters
criteria – query to be used for deleting entries.
- Returns
boolean value indicating whether the deletion has completed successfully.
- async list(options: ~py4ai.data.layer.common.repository.QueryOptions = <py4ai.data.layer.common.repository.QueryOptions object>) Paged[E]
Return a full list of entities stored in the persistence layer.
- Parameters
options – query options to be used when retrieving data
- Returns
Paged object for retrieved list of entities
- async retrieve(key: KE) Optional[E]
Return an entry corresponding to a determined Entity key. If no match is found, returns None.
- Parameters
key – Entity Key to be used for retrieving the entity.
- Returns
Entity associated with the provided key. If no match is found, None is returned.
- async retrieve_by_criteria(criteria: ~py4ai.data.layer.common.criteria.SearchCriteria[~typing.Union[~sqlalchemy.sql.elements.BinaryExpression, ~sqlalchemy.sql.elements.BooleanClauseList]], options: ~py4ai.data.layer.common.repository.QueryOptions = <py4ai.data.layer.common.repository.QueryOptions object>) Paged[E]
Return a list of entities, matching the query provided.
- Parameters
criteria – query to be used for selecting items
options – query options to be used when retrieving data
- Returns
Paged object for retrieved list of entities
- async save(entities: Sequence[E]) Sequence[E]
Create the entries in the persistence layer associated to a list of entities.
- Parameters
entities – list of entities to be created.
- Returns
list of entities that have been successfully created in the persistence layer
- Raises
ValueError – if any object is not inserted correctly
- property serializer: DataSerializer[KE, KD, E, Dict[str, Any]]
Return the data serializer used in the repository.
- Returns
data serializer