sqlalchemy

SQLAlchemy-based implementation of the repository protocol.

SQLAlchemyRepository

SQLAlchemyRepository(*, session, select_=None, **kwargs)

Bases: AbstractRepository[ModelT], Generic[ModelT]

SQLAlchemy based implementation of the repository interface.

select_: To facilitate customization of the underlying select query.

add async

add(data)

Add data to the collection.

Parameters:

Name	Type	Description	Default
data	ModelT	Instance to be added to the collection.	required

Returns:

Type	Description
ModelT	The added instance.

check_health classmethod async

check_health(session)

Perform a health check on the database.

Parameters:

Name	Type	Description	Default
session	AsyncSession	through which we runa check statement	required

Returns:

Type	Description
bool	True if healthy.

delete async

delete(id_)

Delete instance identified by id_.

Parameters:

Name	Type	Description	Default
id_	Any	Identifier of instance to be deleted.	required

Returns:

Type	Description
ModelT	The deleted instance.

Raises:

Type	Description
RepositoryNotFoundException	If no instance found identified by id_.

filter_collection_by_kwargs

filter_collection_by_kwargs(**kwargs)

Filter the collection by kwargs.

Parameters:

Name	Type	Description	Default
**kwargs	Any	key/value pairs such that objects remaining in the collection after filtering have the property that their attribute named key has value equal to value.	{}

get async

get(id_)

Get instance identified by id_.

Parameters:

Name	Type	Description	Default
id_	Any	Identifier of the instance to be retrieved.	required

Returns:

Type	Description
ModelT	The retrieved instance.

Raises:

Type	Description
RepositoryNotFoundException	If no instance found identified by id_.

list async

list(*filters, **kwargs)

Get a list of instances, optionally filtered.

Parameters:

Name	Type	Description	Default
*filters	FilterTypes	Types for specific filtering operations.	()
**kwargs	Any	Instance attribute value filters.	{}

Returns:

Type	Description
list[ModelT]	The list of instances, after filtering applied.

update async

update(data)

Update instance with the attribute values present on data.

Parameters:

Name	Type	Description	Default
data	ModelT	An instance that should have a value for self.id_attribute that exists in the collection.	required

Returns:

Type	Description
ModelT	The updated instance.

Raises:

Type	Description
RepositoryNotFoundException	If no instance found with same identifier as data.

upsert async

upsert(data)

Update or create instance.

Updates instance with the attribute values present on data, or creates a new instance if one doesn't exist.

Parameters:

Name	Type	Description	Default
data	ModelT	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.	required

Returns:

Type	Description
ModelT	The updated or created instance.

Raises:

Type	Description
RepositoryNotFoundException	If no instance found with same identifier as data.

wrap_sqlalchemy_exception

wrap_sqlalchemy_exception()

Do something within context to raise a RepositoryException chained from an original SQLAlchemyError.

>>> try:
...     with wrap_sqlalchemy_exception():
...         raise SQLAlchemyError("Original Exception")
... except StarliteSaqlalchemyError as exc:
...     print(f"caught repository exception from {type(exc.__context__)}")
...
caught repository exception from <class 'sqlalchemy.exc.SQLAlchemyError'>