oxytcmri.interface.repositories.database_repositories

Classes:

Name	Description
DataBaseGateway	Abstract base class for database access to repositories.
DataBaseRepository	Base class for database access to repositories.
DataBaseCenterRepository	Persistence layer for Center entities using a database gateway.
DataBaseAtlasRepository	Persistence layer for Atlas entities using a database gateway.
DataBaseRegionOfInterestRepository	Persistence layer for RegionOfInterest entities using a database gateway.
DataBaseBrainLesionsVolumeRepository	Persistence layer for BrainLesionsVolume entities using a database gateway.
DataBaseMRIExamRepository	Persistence layer for MRIExam entities using a database gateway.
DataBaseSubjectRepository	Persistence layer for Subject entities using a database gateway.
DataBaseDTINormativeValuesRepository	Persistence layer for NormativeValue entities using a database gateway.
DataBaseRepositoriesRegistry

Attributes:

Name	Type	Description
Entity
EntityIdentifier

Entity = TypeVar('Entity') module-attribute

EntityIdentifier = TypeVar('EntityIdentifier') module-attribute

DataBaseGateway

Bases: Generic[Entity], ABC

Abstract base class for database access to repositories.

Methods:

Name	Description
find_by_id	Find an entity by its ID.
find_by_filters	Find an entity by filters. Those filters are the attributes of the entity,
find_all	Find all entities of a given type.
save	Save an entity to the database.
save_list	Save a list of entities to the database.
delete	Delete an entity from the database.
update	Update an entity in the database.
delete_all	Delete all entities of a given type from the database.

find_by_id(entity_type, id_value) abstractmethod

Find an entity by its ID.

Source code in oxytcmri/interface/repositories/database_repositories.py

@abstractmethod
def find_by_id(self, entity_type: Type[Entity], id_value: EntityIdentifier) -> Optional[Entity]:
    """Find an entity by its ID."""

find_by_filters(entity_type, filters) abstractmethod

Find an entity by filters. Those filters are the attributes of the entity, and are modeled as a dictionary.

Source code in oxytcmri/interface/repositories/database_repositories.py

@abstractmethod
def find_by_filters(self, entity_type: Type[Entity], filters: dict[str, Any]) -> Optional[Entity]:
    """Find an entity by filters. Those filters are the attributes of the entity,
    and are modeled as a dictionary."""

find_all(entity_type) abstractmethod

Find all entities of a given type.

Source code in oxytcmri/interface/repositories/database_repositories.py

@abstractmethod
def find_all(self, entity_type: Type[Entity]) -> list[Entity]:
    """Find all entities of a given type."""

save(entity) abstractmethod

Save an entity to the database.

Source code in oxytcmri/interface/repositories/database_repositories.py

@abstractmethod
def save(self, entity: Entity) -> None:
    """Save an entity to the database."""

save_list(entities) abstractmethod

Save a list of entities to the database.

Source code in oxytcmri/interface/repositories/database_repositories.py

@abstractmethod
def save_list(self, entities: List[Entity]) -> None:
    """Save a list of entities to the database."""

delete(entity) abstractmethod

Delete an entity from the database.

Source code in oxytcmri/interface/repositories/database_repositories.py

@abstractmethod
def delete(self, entity: Entity) -> None:
    """Delete an entity from the database."""

update(entity) abstractmethod

Update an entity in the database.

Source code in oxytcmri/interface/repositories/database_repositories.py

@abstractmethod
def update(self, entity: Entity) -> None:
    """Update an entity in the database."""

delete_all(entity_type)

Delete all entities of a given type from the database.

Source code in oxytcmri/interface/repositories/database_repositories.py

def delete_all(self, entity_type: Type[Entity]) -> None:
    """Delete all entities of a given type from the database."""
    for entity in self.find_all(entity_type):
        self.delete(entity)

DataBaseRepository(data_gateway, entity_type, id_extractor)

Bases: Repository[Entity, EntityIdentifier], Generic[Entity, EntityIdentifier]

Base class for database access to repositories.

Initialize the repository with a database gateway.

Parameters:

Name	Type	Description	Default
data_gateway	DataBaseGateway	The database gateway used for accessing the database.	required
entity_type	Type[Entity]	The type of entity managed by this repository.	required
id_extractor	Callable[[Entity], EntityIdentifier]	A function to extract the ID from an entity.	required

Methods:

Name	Description
exists
find_by_id
list_all
save
delete

Attributes:

Name	Type	Description
data_gateway
entity_type

Source code in oxytcmri/interface/repositories/database_repositories.py

def __init__(self,
             data_gateway: DataBaseGateway,
             entity_type: Type[Entity],
             id_extractor: Callable[[Entity], EntityIdentifier]):
    """
    Initialize the repository with a database gateway.

    Parameters
    ----------
    data_gateway : DataBaseGateway
        The database gateway used for accessing the database.
    entity_type : Type[Entity]
        The type of entity managed by this repository.
    id_extractor : Callable[[Entity], EntityIdentifier]
        A function to extract the ID from an entity.
    """
    self.data_gateway = data_gateway
    self.entity_type = entity_type
    self._get_id = id_extractor

data_gateway = data_gateway instance-attribute

entity_type = entity_type instance-attribute

exists(entity)

Source code in oxytcmri/interface/repositories/database_repositories.py

def exists(self, entity: Entity) -> bool:
    entity_id = self._get_id(entity)
    return self.data_gateway.find_by_id(entity_type=self.entity_type, id_value=entity_id) is not None

find_by_id(entity_id)

Source code in oxytcmri/interface/repositories/database_repositories.py

def find_by_id(self, entity_id: EntityIdentifier) -> Optional[Entity]:
    return self.data_gateway.find_by_id(entity_type=self.entity_type,
                                        id_value=entity_id)

list_all()

Source code in oxytcmri/interface/repositories/database_repositories.py

def list_all(self) -> List[Entity]:
    return self.data_gateway.find_all(self.entity_type)

save(entity)

Source code in oxytcmri/interface/repositories/database_repositories.py

def save(self, entity: Entity) -> None:
    self.data_gateway.save(entity)

delete(entity)

Source code in oxytcmri/interface/repositories/database_repositories.py

def delete(self, entity: Entity) -> None:
    self.data_gateway.delete(entity)

DataBaseCenterRepository(data_gateway)

Bases: CenterRepository, DataBaseRepository[Center, int]

Persistence layer for Center entities using a database gateway.

Initialize the repository with a database gateway.

Parameters:

Name	Type	Description	Default
data_gateway	DataBaseGateway	The database gateway used for accessing the database.	required

Source code in oxytcmri/interface/repositories/database_repositories.py

def __init__(self, data_gateway: DataBaseGateway):
    """
    Initialize the repository with a database gateway.

    Parameters
    ----------
    data_gateway : DataBaseGateway
        The database gateway used for accessing the database.
    """
    super().__init__(data_gateway, Center, lambda center: center.id)

DataBaseAtlasRepository(data_gateway)

Bases: AtlasRepository, DataBaseRepository[Atlas, int]

Persistence layer for Atlas entities using a database gateway.

Initialize the repository with a database gateway.

Parameters:

Name	Type	Description	Default
data_gateway	DataBaseGateway	The database gateway used for accessing the database.	required

Source code in oxytcmri/interface/repositories/database_repositories.py

def __init__(self, data_gateway: DataBaseGateway):
    """
    Initialize the repository with a database gateway.

    Parameters
    ----------
    data_gateway : DataBaseGateway
        The database gateway used for accessing the database.
    """
    super().__init__(data_gateway=data_gateway,
                     entity_type=Atlas,
                     id_extractor=lambda atlas: atlas.id)

DataBaseRegionOfInterestRepository(data_gateway)

Bases: RegionOfInterestRepository, DataBaseRepository[RegionOfInterest, str]

Persistence layer for RegionOfInterest entities using a database gateway.

Initialize the repository with a database gateway.

Parameters:

Name	Type	Description	Default
data_gateway	DataBaseGateway	The database gateway used for accessing the database.	required

Source code in oxytcmri/interface/repositories/database_repositories.py

def __init__(self, data_gateway: DataBaseGateway):
    """
    Initialize the repository with a database gateway.

    Parameters
    ----------
    data_gateway : DataBaseGateway
        The database gateway used for accessing the database.
    """
    super().__init__(data_gateway=data_gateway,
                     entity_type=RegionOfInterest,
                     id_extractor=lambda roi: roi.name)

DataBaseBrainLesionsVolumeRepository(data_gateway)

Bases: BrainLesionsVolumeRepository

Persistence layer for BrainLesionsVolume entities using a database gateway.

Initialize the repository with a database gateway.

Parameters:

Name	Type	Description	Default
data_gateway	DataBaseGateway	The database gateway used for accessing the database.	required

Methods:

Name	Description
exists	Check if a BrainLesionsVolume entity exists in the database.
find_by_id
list_all
save
delete	Delete a BrainLesionsVolume entity from the database.

Attributes:

Name	Type	Description
data_gateway

Source code in oxytcmri/interface/repositories/database_repositories.py

def __init__(self, data_gateway: DataBaseGateway):
    """
    Initialize the repository with a database gateway.

    Parameters
    ----------
    data_gateway : DataBaseGateway
        The database gateway used for accessing the database.
    """
    self.data_gateway = data_gateway

data_gateway = data_gateway instance-attribute

exists(entity)

Check if a BrainLesionsVolume entity exists in the database.

Parameters:

Name	Type	Description	Default
entity	BrainLesionsVolume	The BrainLesionsVolume entity to check for existence.	required

Returns:

Type	Description
bool	True if the entity exists, False otherwise.

Source code in oxytcmri/interface/repositories/database_repositories.py

def exists(self, entity: BrainLesionsVolume) -> bool:
    """
    Check if a BrainLesionsVolume entity exists in the database.

    Parameters
    ----------
    entity : BrainLesionsVolume
        The BrainLesionsVolume entity to check for existence.

    Returns
    -------
    bool
        True if the entity exists, False otherwise.
    """
    found_brain_lesions_volume = self.data_gateway.find_by_filters(
        entity_type=BrainLesionsVolume,
        filters={
            'mri_exam_id': entity.mri_exam_id,
            'dti_metric': entity.dti_metric,
            'region_of_interest': entity.region_of_interest,
            'abnormal_value_type': entity.abnormal_value_type
        }
    )
    return found_brain_lesions_volume is not None

find_by_id(entity_id)

Source code in oxytcmri/interface/repositories/database_repositories.py

def find_by_id(self, entity_id: EntityIdentifier) -> Optional[Entity]:
    raise NotImplementedError("find_by_id is not implemented in this Repository")

list_all()

Source code in oxytcmri/interface/repositories/database_repositories.py

def list_all(self) -> List[BrainLesionsVolume]:
    return self.data_gateway.find_all(BrainLesionsVolume)

save(brain_lesions_volume)

Source code in oxytcmri/interface/repositories/database_repositories.py

def save(self, brain_lesions_volume: BrainLesionsVolume) -> None:
    self.data_gateway.save(brain_lesions_volume)

delete(entity)

Delete a BrainLesionsVolume entity from the database.

Parameters:

Name	Type	Description	Default
entity	BrainLesionsVolume	The BrainLesionsVolume entity to delete.	required

Source code in oxytcmri/interface/repositories/database_repositories.py

def delete(self, entity: Entity) -> None:
    """
    Delete a BrainLesionsVolume entity from the database.

    Parameters
    ----------
    entity : BrainLesionsVolume
        The BrainLesionsVolume entity to delete.
    """
    self.data_gateway.delete(entity)

DataBaseMRIExamRepository(data_gateway)

Bases: MRIExamRepository, DataBaseRepository[MRIExam, MRIExamId]

Persistence layer for MRIExam entities using a database gateway.

Initialize the repository with a database gateway.

Parameters:

Name	Type	Description	Default
data_gateway	DataBaseGateway	The database gateway used for accessing the database.	required

Methods:

Name	Description
get_exam_for_subject
save	Save an MRIExam entity to the database, as well as all associated MRIData entities.

Source code in oxytcmri/interface/repositories/database_repositories.py

def __init__(self, data_gateway: DataBaseGateway):
    """
    Initialize the repository with a database gateway.

    Parameters
    ----------
    data_gateway : DataBaseGateway
        The database gateway used for accessing the database.
    """
    super().__init__(
        data_gateway=data_gateway,
        entity_type=MRIExam,
        id_extractor=lambda mri_exam: mri_exam.id
    )

get_exam_for_subject(subject)

Source code in oxytcmri/interface/repositories/database_repositories.py

def get_exam_for_subject(self, subject: Subject) -> MRIExam:
    # Retrieve all MRIExam entities from the database
    all_mri_exams = self.data_gateway.find_all(MRIExam)
    for mri_exam in all_mri_exams:
        if mri_exam.subject_id == subject.id:
            return mri_exam

    # If no MRIExam is found for the given subject_id, raise an exception
    raise LookupError(f"MRIExam with subject_id '{subject_id}' not found.")

save(mri_exam)

Save an MRIExam entity to the database, as well as all associated MRIData entities.

Parameters:

Name	Type	Description	Default
mri_exam	MRIExam	The MRIExam entity to save.	required

Source code in oxytcmri/interface/repositories/database_repositories.py

def save(self, mri_exam: MRIExam) -> None:
    """
    Save an MRIExam entity to the database, as well as all associated MRIData entities.

    Parameters
    ----------
    mri_exam : MRIExam
        The MRIExam entity to save.
    """
    self.data_gateway.save(mri_exam)
    for mri_data in mri_exam.get_all_mri_data():
        self.data_gateway.save(mri_data)

DataBaseSubjectRepository(data_gateway)

Bases: SubjectRepository, DataBaseRepository[Subject, SubjectId]

Persistence layer for Subject entities using a database gateway.

Initialize the repository with a database gateway.

Parameters:

Name	Type	Description	Default
data_gateway	DataBaseGateway	The database gateway used for accessing the database.	required

Source code in oxytcmri/interface/repositories/database_repositories.py

def __init__(self, data_gateway: DataBaseGateway):
    """
    Initialize the repository with a database gateway.

    Parameters
    ----------
    data_gateway : DataBaseGateway
        The database gateway used for accessing the database.
    """
    super().__init__(
        data_gateway=data_gateway,
        entity_type=Subject,
        id_extractor=lambda subject: subject.id
    )

DataBaseDTINormativeValuesRepository(data_gateway)

Bases: NormativeValueRepository

Persistence layer for NormativeValue entities using a database gateway.

Initialize the repository with a database gateway.

Parameters:

Name	Type	Description	Default
data_gateway	DataBaseGateway	The database gateway used for accessing the database.	required

Methods:

Name	Description
save
save_list
exists
get_by_parameters
find_by_id
list_all
delete

Attributes:

Name	Type	Description
data_gateway

Source code in oxytcmri/interface/repositories/database_repositories.py

def __init__(self, data_gateway: DataBaseGateway):
    """
    Initialize the repository with a database gateway.

    Parameters
    ----------
    data_gateway : DataBaseGateway
        The database gateway used for accessing the database.
    """
    self.data_gateway = data_gateway

data_gateway = data_gateway instance-attribute

save(normative_value)

Source code in oxytcmri/interface/repositories/database_repositories.py

def save(self, normative_value: NormativeValue) -> None:
    self.data_gateway.save(normative_value)

save_list(normative_values_list)

Source code in oxytcmri/interface/repositories/database_repositories.py

def save_list(self, normative_values_list: list[NormativeValue]) -> None:
    self.data_gateway.save_list(normative_values_list)

exists(center, dti_metric, atlas, atlas_label, statistic_strategy)

Source code in oxytcmri/interface/repositories/database_repositories.py

def exists(self,
           center: Center,
           dti_metric: DTIMetric,
           atlas: Atlas,
           atlas_label: int,
           statistic_strategy: StatisticStrategy
           ) -> bool:
    normative_values = self.data_gateway.find_by_filters(
        NormativeValue,
        filters={
            'center': center,
            'dti_metric': dti_metric,
            'atlas': atlas,
            'atlas_label': atlas_label,
            'statistic_strategy': statistic_strategy
        }
    )
    return normative_values is not None

get_by_parameters(center, dti_metric, atlas, atlas_label, statistic_strategy)

Source code in oxytcmri/interface/repositories/database_repositories.py

def get_by_parameters(self,
                      center: Center,
                      dti_metric: DTIMetric,
                      atlas: Atlas,
                      atlas_label: int,
                      statistic_strategy: StatisticStrategy
                      ) -> NormativeValue:
    normative_value = self.data_gateway.find_by_filters(
        NormativeValue,
        filters={
            'center': center,
            'dti_metric': dti_metric,
            'atlas': atlas,
            'atlas_label': atlas_label,
            'statistic_strategy': statistic_strategy
        }
    )
    if normative_value is None:
        raise EntityNotFoundException(message=f"NormativeValue with parameters "
                                              f"center = {center}; "
                                              f"DTI metric = {dti_metric}; "
                                              f"atlas = {atlas}; "
                                              f"atlas label = {atlas_label}; "
                                              f"statistic strategy = {statistic_strategy} not found"
                                              f"in repository {self.__class__.__name__}.",
                                      repository=self)
    return normative_value

find_by_id(entity_id)

Source code in oxytcmri/interface/repositories/database_repositories.py

def find_by_id(self, entity_id: int) -> Optional[NormativeValue]:
    raise NotImplementedError("find_by_id is not implemented in this Repository")  # pragma: no cover

list_all()

Source code in oxytcmri/interface/repositories/database_repositories.py

def list_all(self) -> List[NormativeValue]:
    return self.data_gateway.find_all(NormativeValue)

delete(entity)

Source code in oxytcmri/interface/repositories/database_repositories.py

def delete(self, entity: NormativeValue) -> None:
    self.data_gateway.delete(entity)

DataBaseRepositoriesRegistry(persistence_gateway)

Bases: RepositoriesRegistry

Methods:

Name	Description
get_repository
list_all_repositories

Source code in oxytcmri/interface/repositories/database_repositories.py

def __init__(self, persistence_gateway: DataBaseGateway):
    self._repositories = {
        Center: DataBaseCenterRepository(persistence_gateway),
        Subject: DataBaseSubjectRepository(persistence_gateway),
        MRIExam: DataBaseMRIExamRepository(persistence_gateway),
        Atlas: DataBaseAtlasRepository(persistence_gateway),
        NormativeValue: DataBaseDTINormativeValuesRepository(persistence_gateway),
        RegionOfInterest: DataBaseRegionOfInterestRepository(persistence_gateway),
        BrainLesionsVolume: DataBaseBrainLesionsVolumeRepository(persistence_gateway),
    }

get_repository(entity_type)

Source code in oxytcmri/interface/repositories/database_repositories.py

def get_repository(self, entity_type: Type[Entity]) -> "Repository[Entity, Any]":
    repository = self._repositories.get(entity_type)
    if repository is None:
        raise ValueError(f"No repository registered for entity type: {entity_type}")
    return repository

list_all_repositories()

Source code in oxytcmri/interface/repositories/database_repositories.py

def list_all_repositories(self) -> List[Repository[Entity, Any]]:
    return list(self._repositories.values())