Persistence Layer

Every persistence service is implementing one of interface defined in this module. Interfaces are generics depends on at least Entity class (finders needs also request class).

Example:Implementing of persistence service

>>> from acd.domain import Entity
...
>>> class Example(Entity, id_attributes=('num',)):
...     num: int
...
>>> class ExampleInserter(Inserter[Example]):
...     def __init__(self, dbsession):
...         self._session = dbsession
...
...     def __call__(self, entity: Example) -> Example:
...         self._session.add(entity)
...         return entity

Inserter

class acd.persistence.Inserter

Stores entity.

Example:Inserter implementation

>>> from acd.domain import Entity
>>> class Example(Entity, id_attributes=('num',)):
...    num: int
...
>>> class ExampleInserter(Inserter[Example]):
...     def __init__(self, cache: dict) -> None:
...         self._cache = cache
...
...     def __call__(self, entity: Example) -> Example:
...         if entity.entity_id in self._cache:
...             raise EntityDuplicateError(entity.entity_id)
...         self._cache[entity.entity_id] = entity
...         return entity
...
>>> insert_example = ExampleInserter({})
>>> insert_example(Example(num=1))
Example(num=1)

class acd.persistence.EntityDuplicateError

Raises when inserting duplicate of entity.

Updater

class acd.persistence.Updater

Updates stored entity with new entity version.

Example:Updater implementation

>>> from acd.domain import Entity
>>> class Example(Entity, id_attributes=('num',)):
...    num: int
...
>>> class ExampleUpdater(Updater[Example]):
...     def __init__(self, cache: dict) -> None:
...         self._cache = cache
...
...     def __call__(self, entity: Example) -> Example:
...         if entity.entity_id not in self._cache:
...             raise EntityNotFoundError(entity.entity_id)
...         self._cache[entity.entity_id] = entity
...         return entity
...
>>> entity = Example(num=1)
>>> cache = {entity.entity_id: entity}
>>> update_example = ExampleUpdater(cache)
>>> update_example(Example(num=1))
Example(num=1)

class acd.persistence.EntityNotFoundError

Raises when trying to update not existing entity.

Getter

class acd.persistence.Getter

Gets stored entity.

Example:Getter implementation

>>> from acd.domain import Entity
>>> class Example(Entity, id_attributes=('num',)):
...    num: int
...
>>> class ExampleGetter(Getter[Example]):
...     def __init__(self, cache: dict) -> None:
...         self._cache = cache
...
...     def __call__(self, entity_id: EntityId) -> Optional[Example]:
...         return self._cache.get(entity_id)
...
>>> entity = Example(num=1)
>>> cache = {entity.entity_id: entity}
>>> get_example = ExampleGetter(cache)
>>> get_example((1,))
Example(num=1)

Finder

class acd.persistence.Finder

Finds all entities for given search request.

Example:Finder implementation

>>> from acd.domain import Entity
>>> from acd.libs.annotation import Immutable
...
>>> class Example(Entity, id_attributes=('num',)):
...     num: int
...
>>> class RangeSearchRequest(Immutable):
...     greater_then: int
...     lower_then: int
...
>>> class ExampleFinder(Finder[RangeSearchRequest, Example]):
...     def __init__(self, cache: dict) -> None:
...         self._cache = cache
...
...     def __call__(self, request: RangeSearchRequest) -> List[Example]:
...         return [
...             e for e in self._cache.values()
...             if request.greater_then < e.num < request.lower_then
...         ]
...
>>> entities = [
...     Example(num=1), Example(num=2), Example(num=3), Example(num=4),
... ]
>>> cache = {e.entity_id: e for e in entities}
>>> find_examples = ExampleFinder(cache)
>>> request = RangeSearchRequest(greater_then=1, lower_then=4)
>>> find_examples(request)
[Example(num=2), Example(num=3)]