Skip to content

oxytcmri.domain.ports.repositories

This module defines repositories for the application, storing and retrieving entities.

Classes:

Name Description
EntityNotFoundException

Generic exception raised when an entity is not found in the repository.
EntityIdNotFoundException

Exception raised when an entity is not found based on its ID.
Repository

Abstract base class for repositories.
SubjectRepository
MRIExamRepository

Abstract base class for MRI repository.
AtlasRepository

Abstract base class for Atlas repository.
RegionOfInterestRepository

Abstract base class for Region of Interest repository.
CenterRepository

Abstract base class for Center repository.
RepositoriesRegistry

Abstract base class for a registry responsible for managing the repositories in the application.

Attributes:

Name Type Description
Entity
EntityIdentifier

Entity = TypeVar('Entity') module-attribute

EntityIdentifier = TypeVar('EntityIdentifier') module-attribute

EntityNotFoundException(message, repository)

Bases: LookupError

Generic exception raised when an entity is not found in the repository.

Attributes:

Name Type Description
repository
Source code in oxytcmri/domain/ports/repositories.py 
22
23
24
def __init__(self, message: str, repository: Repository):
    super().__init__(message)
    self.repository = repository

repository = repository instance-attribute

EntityIdNotFoundException(entity_id, repository)

Bases: EntityNotFoundException

Exception raised when an entity is not found based on its ID.

Attributes:

Name Type Description
entity_id
Source code in oxytcmri/domain/ports/repositories.py 
32
33
34
35
def __init__(self, entity_id: EntityIdentifier, repository: Repository):
    message = f"Entity with ID {entity_id} not found in repository {repository}"
    super().__init__(message=message, repository=repository)
    self.entity_id = entity_id

entity_id = entity_id instance-attribute

Repository

Bases: ABC, Generic[Entity, EntityIdentifier]

Abstract base class for repositories. Defines the interface for all repositories in the application.

Methods:

Name Description
exists

Check if an entity exists in the repository
find_by_id

Retrieve an entity by its ID.
get_by_id

Retrieve an entity by its ID.
list_all

List all entities in the repository.
save

Save an entity to the repository.
save_list

Save a list of entities to the repository.
delete

Delete an entity from the repository.
update

Update an existing entity in the repository, meaning it will first delete it if it exists and then save the

exists(entity) abstractmethod

Check if an entity exists in the repository

Parameters:

Name Type Description Default
entity Entity

The entity to check for existence

 required

Returns:

Type Description
bool

True if the entity exists, False otherwise
Source code in oxytcmri/domain/ports/repositories.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
@abstractmethod
def exists(self, entity: Entity) -> bool:
    """
    Check if an entity exists in the repository

    Parameters
    ----------
    entity : Entity
        The entity to check for existence

    Returns
    -------
    bool
        True if the entity exists, False otherwise
    """

find_by_id(entity_id) abstractmethod

Retrieve an entity by its ID.

Parameters:

Name Type Description Default
entity_id EntityIdentifier

The ID of the entity

 required

Returns:

Type Description
Entity

The entity object if found, otherwise None
Source code in oxytcmri/domain/ports/repositories.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@abstractmethod
def find_by_id(self, entity_id: EntityIdentifier) -> Optional[Entity]:
    """
    Retrieve an entity by its ID.

    Parameters
    ----------
    entity_id : EntityIdentifier
        The ID of the entity

    Returns
    -------
    Entity
        The entity object if found, otherwise None
    """

get_by_id(entity_id)

Retrieve an entity by its ID.

Parameters:

Name Type Description Default
entity_id EntityIdentifier

The ID of the entity

 required

Returns:

Type Description
Entity

The entity object

Raises:

Type Description
EntityIdNotFoundException

If the entity with the given ID does not exist
Source code in oxytcmri/domain/ports/repositories.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def get_by_id(self, entity_id: EntityIdentifier) -> Entity:
    """
    Retrieve an entity by its ID.

    Parameters
    ----------
    entity_id : EntityIdentifier
        The ID of the entity

    Returns
    -------
    Entity
        The entity object

    Raises
    -------
    EntityIdNotFoundException
        If the entity with the given ID does not exist
    """
    entity = self.find_by_id(entity_id)
    if entity is None:
        raise EntityIdNotFoundException(entity_id, self)
    return entity

list_all() abstractmethod

List all entities in the repository.

Returns:

Type Description
List[Entity]

List of all entities
Source code in oxytcmri/domain/ports/repositories.py 
100
101
102
103
104
105
106
107
108
109
@abstractmethod
def list_all(self) -> List[Entity]:
    """
    List all entities in the repository.

    Returns
    -------
    List[Entity]
        List of all entities
    """

save(entity) abstractmethod

Save an entity to the repository.

Parameters:

Name Type Description Default
entity Entity

The entity to save

 required
Source code in oxytcmri/domain/ports/repositories.py 
111
112
113
114
115
116
117
118
119
120
@abstractmethod
def save(self, entity: Entity) -> None:
    """
    Save an entity to the repository.

    Parameters
    ----------
    entity : Entity
        The entity to save
    """

save_list(entity_list)

Save a list of entities to the repository.

Parameters:

Name Type Description Default
entity_list List[Entity]

The list of entities to save

 required
Source code in oxytcmri/domain/ports/repositories.py 
122
123
124
125
126
127
128
129
130
131
132
def save_list(self, entity_list: List[Entity]) -> None:
    """
    Save a list of entities to the repository.

    Parameters
    ----------
    entity_list : List[Entity]
        The list of entities to save
    """
    for entity in entity_list:
        self.save(entity)

delete(entity) abstractmethod

Delete an entity from the repository.

Parameters:

Name Type Description Default
entity Entity

The entity to delete

 required
Source code in oxytcmri/domain/ports/repositories.py 
134
135
136
137
138
139
140
141
142
143
@abstractmethod
def delete(self, entity: Entity) -> None:
    """
    Delete an entity from the repository.

    Parameters
    ----------
    entity : Entity
        The entity to delete
    """

update(entity)

Update an existing entity in the repository, meaning it will first delete it if it exists and then save the updated entity.

Parameters:

Name Type Description Default
entity Entity

The entity to update

 required
Source code in oxytcmri/domain/ports/repositories.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
def update(self, entity: Entity) -> None:
    """
    Update an existing entity in the repository, meaning it will first delete it if it exists and then save the
    updated entity.

    Parameters
    ----------
    entity : Entity
        The entity to update
    """
    if self.exists(entity):
        self.delete(entity)
    self.save(entity)

SubjectRepository

Bases: Repository[Subject, SubjectId], ABC

Methods:

Name Description
find_subjects_by_center

Find subjects for a given center, optionally filtered by type.
list_all_patients

List all patients in the repository.

find_subjects_by_center(center, subject_type=None)

Find subjects for a given center, optionally filtered by type.

Parameters:

Name Type Description Default
center Center

The center to find subjects for

 required
subject_type Optional[SubjectType]

If provided, only return subjects of this type

 None

Returns:

Type Description
List[Subject]

List of matching subjects
Source code in oxytcmri/domain/ports/repositories.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def find_subjects_by_center(
        self, center: Center, subject_type: Optional[SubjectType] = None
) -> List[Subject]:
    """
    Find subjects for a given center, optionally filtered by type.

    Parameters
    ----------
    center : Center
        The center to find subjects for
    subject_type : Optional[SubjectType], default=None
        If provided, only return subjects of this type

    Returns
    -------
    List[Subject]
        List of matching subjects
    """
    all_subjects = self.list_all()
    results = []

    for subject in all_subjects:
        # filter by center
        if subject.center_id == center.id:
            # filter by subject type
            if subject_type is None or subject.subject_type == subject_type:
                results.append(subject)

    return results

list_all_patients()

List all patients in the repository.

Returns:

Type Description
List[Subject]

List of all patients
Source code in oxytcmri/domain/ports/repositories.py 
192
193
194
195
196
197
198
199
200
201
202
def list_all_patients(self) -> List[Subject]:
    """
    List all patients in the repository.

    Returns
    -------
    List[Subject]
        List of all patients
    """
    all_subjects = self.list_all()
    return [subject for subject in all_subjects if subject.subject_type == SubjectType.PATIENT]

MRIExamRepository

Bases: Repository[MRIExam, MRIExamId]

Abstract base class for MRI repository. Defines the interface for retrieving MRI exam data.

Methods:

Name Description
get_exam_for_subject

Retrieve the MRI exam for a specific subject.

get_exam_for_subject(subject) abstractmethod

Retrieve the MRI exam for a specific subject.

Parameters:

Name Type Description Default
subject Subject

The subject to retrieve the MRI exam for

 required

Returns:

Type Description
MRIExam

The MRI exam for the subject
Source code in oxytcmri/domain/ports/repositories.py 
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
@abstractmethod
def get_exam_for_subject(self, subject: Subject) -> MRIExam:
    """
    Retrieve the MRI exam for a specific subject.

    Parameters
    ----------
    subject : Subject
        The subject to retrieve the MRI exam for

    Returns
    -------
    MRIExam
        The MRI exam for the subject
    """

AtlasRepository

Bases: Repository[Atlas, int], ABC

Abstract base class for Atlas repository. Defines the interface for retrieving atlas data.

RegionOfInterestRepository

Bases: Repository[RegionOfInterest, str], ABC

Abstract base class for Region of Interest repository. Defines the interface for retrieving regions of interest data.

CenterRepository

Bases: Repository[Center, int], ABC

Abstract base class for Center repository.

Methods:

Name Description
get_by_mri_exam_id

Retrieve the center associated with a specific MRI exam ID.

get_by_mri_exam_id(mri_exam_id)

Retrieve the center associated with a specific MRI exam ID.

Parameters:

Name Type Description Default
mri_exam_id MRIExamId

The ID of the MRI exam

 required

Returns:

Type Description
Center

The center associated with the MRI exam ID

Raises:

Type Description
EntityNotFoundException

If the center with the given MRIExamId does not exist
Source code in oxytcmri/domain/ports/repositories.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
def get_by_mri_exam_id(self, mri_exam_id: MRIExamId) -> Center:
    """
    Retrieve the center associated with a specific MRI exam ID.

    Parameters
    ----------
    mri_exam_id : MRIExamId
        The ID of the MRI exam

    Returns
    -------
    Center
        The center associated with the MRI exam ID

    Raises
    ------
    EntityNotFoundException
        If the center with the given MRIExamId does not exist
    """
    subject_id: SubjectId = mri_exam_id.to_subject_id()
    center_id = subject_id.center_id
    center = self.find_by_id(center_id)
    if center is None:
        raise EntityNotFoundException(
            f"Center with ID {center_id} not found for MRI exam ID {mri_exam_id}",
            self,
        )
    return center

RepositoriesRegistry

Bases: ABC

Abstract base class for a registry responsible for managing the repositories in the application.

Methods:

Name Description
get_repository

Return the repository corresponding to the given entity type.
list_all_repositories

Return a list of all repositories in the registry.

get_repository(entity_type) abstractmethod

Return the repository corresponding to the given entity type.

Source code in oxytcmri/domain/ports/repositories.py 
278
279
280
@abstractmethod
def get_repository(self, entity_type: Type[Entity]) -> Repository[Entity, EntityIdentifier]:
    """Return the repository corresponding to the given entity type."""

list_all_repositories() abstractmethod

Return a list of all repositories in the registry.

Source code in oxytcmri/domain/ports/repositories.py 
282
283
284
@abstractmethod
def list_all_repositories(self) -> List[Repository[Entity, EntityIdentifier]]:
    """Return a list of all repositories in the registry."""