"""
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,
Panel,
PanelDecision,
PanelReview,
Project,
Proposal,
ProposalAccess,
SBDefinition,
SBInstance,
)
from ska_oso_pdm.entity_status_history import (
OSOEBStatusHistory,
ProjectStatusHistory,
SBDStatusHistory,
SBIStatusHistory,
)
from ska_oso_pdm.sb_definition.sb_definition import SBDefinitionID
from ska_db_oda.persistence.domain.metadatamixin import MetadataMixin
from ska_db_oda.persistence.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 ODAError: 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 NotFoundInODA: if the sbd_id is not found in the repository
:raises ODAError: if an error occurs while retrieving the SBD
"""
raise NotImplementedError
[docs]
@abstractmethod
def read_relationship(
self, entity_id: U, parent_entity: U, associated_entity: U
) -> T:
"""
Retrieves the entity data associated with the given parent entity ID from the ODA.
:raises NotFoundInODA: if the sbd_id is not found in the repository
:raises ODAError: 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 ODAError: 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 QueryParameterError: if the qry_params are not supported
:raises ODAError: 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.
It is expected to be extended as 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, user: str = None) -> 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 ODAError: 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,
ProposalAccess,
PanelDecision,
PanelReview,
Panel,
ProposalAccess,
SBDStatusHistory,
SBIStatusHistory,
OSOEBStatusHistory,
ProjectStatusHistory,
),
):
return self._bridge.create(entity, user)
elif isinstance(entity, (OSOExecutionBlock, SBInstance)):
return self._bridge.update(entity)
[docs]
def get(
self,
entity_id: U,
version: U = None,
) -> T:
"""Retrieves the latest version of the entity with the given id from the ODA.
:param entity_id: entity_id of the entity
:param version: version of the entity
:raises NotFoundInODA: if the entity_id is not found in the repository
:raises ODAError: if an error occurs while retrieving the entity
"""
return self._bridge.read(entity_id, version)
[docs]
def get_relationship(
self, entity_id: U, parent_entity: U, associated_entity: U
) -> T:
"""Retrieves the entity data associated with the given parent entity ID from the ODA.
:raises NotFoundInODA: if the sbd_id is not found in the repository
:raises ODAError: if an error occurs while retrieving the SBD
:param entity_id: entity id to be search.
:param parent_entity: primary entity name to be join.
:param associated_entity: secondary entity name to be join.
"""
return self._bridge.read_relationship(
entity_id, parent_entity, associated_entity
)
[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 ODAError: 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 SBDefinitionStatusHistoryRepository(AbstractRepository[SBDStatusHistory, str]):
"""
Abstraction over persistent storage of SBDefinitions
"""
[docs]
class ExecutionBlockRepository(AbstractRepository[OSOExecutionBlock, str]):
"""
Abstraction over persistent storage of OSOExecutionBlocks
"""
[docs]
class ExecutionBlockStatusHistoryRepository(
AbstractRepository[OSOEBStatusHistory, str]
):
"""
Abstraction over persistent storage of ExecutionBlock
"""
[docs]
class ProjectRepository(AbstractRepository[Project, str]):
"""
Abstraction over persistent storage of Projects
"""
[docs]
class ProjectStatusHistoryRepository(AbstractRepository[ProjectStatusHistory, str]):
"""
Abstraction over persistent storage of Projects
"""
[docs]
class SBInstanceRepository(AbstractRepository[SBInstance, str]):
"""
Abstraction over persistent storage of SBInstances
"""
[docs]
class SBInstanceStatusHistoryRepository(AbstractRepository[SBIStatusHistory, str]):
"""
Abstraction over persistent storage of SBInstance
"""
[docs]
class ProposalRepository(AbstractRepository[Proposal, str]):
"""
Abstraction over persistent storage of Proposals
"""
[docs]
class PanelDecisionRepository(AbstractRepository[PanelDecision, str]):
"""
Abstraction over persistent storage of PanelDecisions
"""
[docs]
class PanelReviewRepository(AbstractRepository[PanelReview, str]):
"""
Abstraction over persistent storage of PanelReviews
"""
[docs]
class PanelRepository(AbstractRepository[Panel, str]):
"""
Abstraction over persistent storage of PM Panels
"""
[docs]
class ProposalAccessRepository(AbstractRepository[ProposalAccess, str]):
"""
Abstraction over persistent storage of Proposal Access
"""
[docs]
class EntityRelationshipRepository(AbstractRepository[any, any]):
"""
Abstraction over persistent storage of Relational Entities
"""