"""
This module contains the AbstractRepository base class and the entity specific types.
"""
from abc import abstractmethod
from typing import Generic, List, TypeVar
from ska_oso_pdm import OSOExecutionBlock, Project, Proposal, SBDefinition, SBInstance
from ska_oso_pdm.sb_definition.sb_definition import SBDefinitionID
from ska_db_oda.domain.metadatamixin import MetadataMixin
from ska_db_oda.domain.query import QueryParams
T = TypeVar("T")
U = TypeVar("U")
[docs]
class RepositoryBridge(Generic[T, U], MetadataMixin[T]):
"""
This class is the implementor of the Bridge pattern which decouples the persistence method from the Repository abstraction.
It is designed to be used as a composition within a repository and offers CRUD type methods.
"""
[docs]
@abstractmethod
def create(self, entity: T) -> T:
"""
Stores a new, versioned entity in the repository
:raises ValueError: if the validation of the entity or its metadata fails
:raises OSError: if an error occurs while persisting the entity
:return: the entity as it exists in the ODA
"""
raise NotImplementedError
[docs]
@abstractmethod
def read(self, entity_id: U) -> T:
"""
Retrieves the latest version of the entity with the given id from the ODA.
:raises KeyError: if the sbd_id is not found in the repository
:raises OSError: if an error occurs while retrieving the SBD
"""
raise NotImplementedError
[docs]
@abstractmethod
def update(self, entity: T) -> T:
"""
Updates version 1 of the entity with the given entity ID in the repository, or creates
version 1 if it doesn't already exist.
:raises ValueError: if the validation of the entity or its metadata fails
:raises OSError: if an error occurs while persisting the entity
:return: the entity as it exists in the ODA
"""
raise NotImplementedError
[docs]
@abstractmethod
def query(self, qry_params: QueryParams) -> List[U]:
"""Queries the latest version of the entity based on QueryParams class from the ODA
and returns the corresponding entity ID
Returns an empty list if no entities in the repository match the parameters.
:raises ValueError: if the qry_params are not supported
:raises OSError: if an error occurs while querying the entity
"""
raise NotImplementedError
@abstractmethod
def __contains__(self, item):
raise NotImplementedError
@abstractmethod
def __len__(self):
raise NotImplementedError
[docs]
class AbstractRepository(Generic[T, U]):
"""
Generic repository that defines the interface for users to add and retrieve entities from the ODA.
The implementation is passed at runtime (eg postgres, filesystem) and metadata
updates are handled via the mixin class.
It is expected to be typed to SBDefinitionRepository, SBInstanceRepository, etc.
"""
def __init__(self, bridge: RepositoryBridge[T, U]):
self._bridge = bridge
def __contains__(self, entity_id: U):
return entity_id in self._bridge
def __len__(self):
return len(self._bridge)
[docs]
def add(self, entity: T) -> T:
"""Stores the entity in the ODA.
The entity passed to this method will have its metadata validated and updated.
:raises ValueError: if the validation of the sbd or its metadata fails
:raises OSError: if an error occurs while persisting the entity
:return: the entity as it exists in the ODA, eg with updated metadata
"""
if isinstance(entity, (SBDefinition, Project, Proposal)):
return self._bridge.create(entity)
elif isinstance(entity, (OSOExecutionBlock, SBInstance)):
return self._bridge.update(entity)
[docs]
def get(self, entity_id: U) -> T:
"""Retrieves the latest version of the entity with the given id from the ODA.
:raises KeyError: if the sbd_id is not found in the repository
:raises OSError: if an error occurs while retrieving the SBD
"""
return self._bridge.read(entity_id)
[docs]
def query(self, qry_param: QueryParams) -> List[T]:
"""Queries the latest version of the entity based on QueryParams class from the ODA
and returns the corresponding entities
Returns an empty list if no entities in the repository match the parameters.
:raises ValueError: if the qry_params are not supported
:raises OSError: if an error occurs while querying the entity
"""
return self._bridge.query(qry_param)
[docs]
class SBDefinitionRepository(AbstractRepository[SBDefinition, SBDefinitionID]):
"""
Abstraction over persistent storage of SBDefinitions
"""
[docs]
class ExecutionBlockRepository(AbstractRepository[OSOExecutionBlock, str]):
"""
Abstraction over persistent storage of OSOExecutionBlocks
"""
[docs]
class ProjectRepository(AbstractRepository[Project, str]):
"""
Abstraction over persistent storage of Projects
"""
[docs]
class SBInstanceRepository(AbstractRepository[SBInstance, str]):
"""
Abstraction over persistent storage of SBInstances
"""
[docs]
class ProposalRepository(AbstractRepository[Proposal, str]):
"""
Abstraction over persistent storage of Proposals
"""