Skip to content

db_entity

This module defines the DatabaseEntity abstract base class, which serves as a common interface for interacting with database entities such as studies, runs, and workers. The DatabaseEntity class provides a standardized structure for managing these entities, including methods for saving, deleting, reloading, and retrieving additional metadata.

Classes that inherit from DatabaseEntity must implement the abstract methods defined in the base class, ensuring consistency across different types of database entities. This abstraction reduces code duplication and promotes maintainability by centralizing shared functionality.

DatabaseEntity

Bases: Generic[T], ABC

Abstract base class for database entities such as runs, studies, and workers.

This class defines the common interface and behavior for interacting with database entities, including saving, deleting, and reloading data.

Attributes:

Name Type Description
entity_info T

The data model containing information about the entity.
backend ResultsBackend

The backend instance used to interact with the database.

Methods:

Name Description
__repr__

Provide a string representation of the entity.
__str__

Provide a human-readable string representation of the entity.
reload_data

Reload the latest data for this entity from the database.
get_id

Get the unique ID for this entity.
get_additional_data

Get any additional data saved to this entity.
save

Save the current state of this entity to the database.
_post_save_hook

Hook called after saving the entity to the database. Subclasses can override this method to add additional behavior.
load

Load an entity from the database by its ID.
delete

Delete an entity from the database by its ID.
entity_type

Get the type of this entity for database operations.
_get_entity_type

Get the entity type for database operations (used internally).
Source code in merlin/db_scripts/entities/db_entity.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
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
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
class DatabaseEntity(Generic[T], ABC):
    """
    Abstract base class for database entities such as runs, studies, and workers.

    This class defines the common interface and behavior for interacting with
    database entities, including saving, deleting, and reloading data.

    Attributes:
        entity_info (T): The data model containing information about the entity.
        backend (backends.results_backend.ResultsBackend): The backend instance used
            to interact with the database.

    Methods:
        __repr__:
            Provide a string representation of the entity.

        __str__:
            Provide a human-readable string representation of the entity.

        reload_data:
            Reload the latest data for this entity from the database.

        get_id:
            Get the unique ID for this entity.

        get_additional_data:
            Get any additional data saved to this entity.

        save:
            Save the current state of this entity to the database.

        _post_save_hook:
            Hook called after saving the entity to the database. Subclasses can override
            this method to add additional behavior.

        load:
            Load an entity from the database by its ID.

        delete:
            Delete an entity from the database by its ID.

        entity_type:
            Get the type of this entity for database operations.

        _get_entity_type:
            Get the entity type for database operations (used internally).
    """

    # Mapping of entity types to their error classes
    _error_classes = {
        "logical_worker": WorkerNotFoundError,
        "physical_worker": WorkerNotFoundError,
        "run": RunNotFoundError,
        "study": StudyNotFoundError,
    }

    def __init__(self, entity_info: T, backend: ResultsBackend):
        """
        Initialize a `DatabaseEntity` instance.

        Args:
            entity_info (T): The data model containing information about the entity.
            backend (backends.results_backend.ResultsBackend): The backend instance used to
                interact with the database.
        """
        self.entity_info: T = entity_info
        self.backend: ResultsBackend = backend

    @abstractmethod
    def __repr__(self) -> str:
        """
        Provide a string representation of the entity.
        """
        raise NotImplementedError("Subclasses of `DatabaseEntity` must implement a `__repr__` method.")

    @abstractmethod
    def __str__(self) -> str:
        """
        Provide a human-readable string representation of the entity.
        """
        raise NotImplementedError("Subclasses of `DatabaseEntity` must implement a `__str__` method.")

    @property
    def entity_type(self) -> str:
        """
        Get the type of this entity for database operations.

        Returns:
            The type name as a string.
        """
        return self._get_entity_type()

    @classmethod
    def _get_entity_type(cls) -> str:
        """
        Get the entity type for database operations.

        Returns:
            The type name as a string.
        """
        # Default implementation based on class name
        # Can be overridden by subclasses if needed
        return cls.__name__.lower().replace("entity", "")

    def reload_data(self):
        """
        Reload the latest data for this entity from the database.

        Raises:
            (exceptions.EntityNotFoundError): If an entry for this entity was not found in the database.
        """
        entity_id = self.get_id()
        updated_entity_info = self.backend.retrieve(entity_id, self.entity_type)
        if not updated_entity_info:
            error_class = self._error_classes.get(self.entity_type, EntityNotFoundError)
            raise error_class(f"{self.entity_type.capitalize()} with ID {entity_id} not found in the database.")
        self.entity_info = updated_entity_info

    def get_id(self) -> str:
        """
        Get the unique ID for this entity.

        Returns:
            The unique ID for this entity.
        """
        return self.entity_info.id

    def get_additional_data(self) -> Dict:
        """
        Get any additional data saved to this entity.

        Returns:
            A dictionary of additional data.
        """
        self.reload_data()
        return self.entity_info.additional_data

    def save(self):
        """
        Save the current state of this entity to the database.
        """
        self.backend.save(self.entity_info)
        self._post_save_hook()

    def _post_save_hook(self):
        """
        Hook called after saving the entity to the database.
        Subclasses can override to add additional behavior.
        """

    @classmethod
    def load(cls: Type[E], entity_identifier: str, backend: ResultsBackend) -> E:
        """
        Load an entity from the database by its ID.

        Args:
            entity_identifier: The ID of the entity to load.
            backend (backends.results_backend.ResultsBackend): A `ResultsBackend` instance.

        Returns:
            An instance of the entity.

        Raises:
            (exceptions.EntityNotFoundError): If an entry for the entity was not found in the database.
        """
        entity_info = backend.retrieve(entity_identifier, cls._get_entity_type())
        if not entity_info:
            error_class = cls._error_classes.get(cls._get_entity_type(), EntityNotFoundError)
            raise error_class(f"{cls._get_entity_type().capitalize()} with ID {entity_identifier} not found in the database.")

        return cls(entity_info, backend)

    @classmethod
    def delete(cls, entity_identifier: str, backend: ResultsBackend):
        """
        Delete an entity from the database by its ID.

        Args:
            entity_identifier: The ID of the entity to delete.
            backend (backends.results_backend.ResultsBackend): A ResultsBackend instance.
        """
        entity_type = cls._get_entity_type()
        LOG.debug(f"Deleting {entity_type} with ID '{entity_identifier}' from the database...")
        backend.delete(entity_identifier, entity_type)
        LOG.info(f"{entity_type.capitalize()} with ID '{entity_identifier}' has been successfully deleted.")

entity_type property

Get the type of this entity for database operations.

Returns:

Type Description
str

The type name as a string.

__init__(entity_info, backend)

Initialize a DatabaseEntity instance.

Parameters:

Name Type Description Default
entity_info T

The data model containing information about the entity.

 required
backend ResultsBackend

The backend instance used to interact with the database.

 required
Source code in merlin/db_scripts/entities/db_entity.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def __init__(self, entity_info: T, backend: ResultsBackend):
    """
    Initialize a `DatabaseEntity` instance.

    Args:
        entity_info (T): The data model containing information about the entity.
        backend (backends.results_backend.ResultsBackend): The backend instance used to
            interact with the database.
    """
    self.entity_info: T = entity_info
    self.backend: ResultsBackend = backend

__repr__() abstractmethod

Provide a string representation of the entity.

Source code in merlin/db_scripts/entities/db_entity.py 
103
104
105
106
107
108
@abstractmethod
def __repr__(self) -> str:
    """
    Provide a string representation of the entity.
    """
    raise NotImplementedError("Subclasses of `DatabaseEntity` must implement a `__repr__` method.")

__str__() abstractmethod

Provide a human-readable string representation of the entity.

Source code in merlin/db_scripts/entities/db_entity.py 
110
111
112
113
114
115
@abstractmethod
def __str__(self) -> str:
    """
    Provide a human-readable string representation of the entity.
    """
    raise NotImplementedError("Subclasses of `DatabaseEntity` must implement a `__str__` method.")

delete(entity_identifier, backend) classmethod

Delete an entity from the database by its ID.

Parameters:

Name Type Description Default
entity_identifier str

The ID of the entity to delete.

 required
backend ResultsBackend

A ResultsBackend instance.

 required
Source code in merlin/db_scripts/entities/db_entity.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
@classmethod
def delete(cls, entity_identifier: str, backend: ResultsBackend):
    """
    Delete an entity from the database by its ID.

    Args:
        entity_identifier: The ID of the entity to delete.
        backend (backends.results_backend.ResultsBackend): A ResultsBackend instance.
    """
    entity_type = cls._get_entity_type()
    LOG.debug(f"Deleting {entity_type} with ID '{entity_identifier}' from the database...")
    backend.delete(entity_identifier, entity_type)
    LOG.info(f"{entity_type.capitalize()} with ID '{entity_identifier}' has been successfully deleted.")

get_additional_data()

Get any additional data saved to this entity.

Returns:

Type Description
Dict

A dictionary of additional data.
Source code in merlin/db_scripts/entities/db_entity.py 
162
163
164
165
166
167
168
169
170
def get_additional_data(self) -> Dict:
    """
    Get any additional data saved to this entity.

    Returns:
        A dictionary of additional data.
    """
    self.reload_data()
    return self.entity_info.additional_data

get_id()

Get the unique ID for this entity.

Returns:

Type Description
str

The unique ID for this entity.
Source code in merlin/db_scripts/entities/db_entity.py 
153
154
155
156
157
158
159
160
def get_id(self) -> str:
    """
    Get the unique ID for this entity.

    Returns:
        The unique ID for this entity.
    """
    return self.entity_info.id

load(entity_identifier, backend) classmethod

Load an entity from the database by its ID.

Parameters:

Name Type Description Default
entity_identifier str

The ID of the entity to load.

 required
backend ResultsBackend

A ResultsBackend instance.

 required

Returns:

Type Description
E

An instance of the entity.

Raises:

Type Description
EntityNotFoundError

If an entry for the entity was not found in the database.
Source code in merlin/db_scripts/entities/db_entity.py 
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
@classmethod
def load(cls: Type[E], entity_identifier: str, backend: ResultsBackend) -> E:
    """
    Load an entity from the database by its ID.

    Args:
        entity_identifier: The ID of the entity to load.
        backend (backends.results_backend.ResultsBackend): A `ResultsBackend` instance.

    Returns:
        An instance of the entity.

    Raises:
        (exceptions.EntityNotFoundError): If an entry for the entity was not found in the database.
    """
    entity_info = backend.retrieve(entity_identifier, cls._get_entity_type())
    if not entity_info:
        error_class = cls._error_classes.get(cls._get_entity_type(), EntityNotFoundError)
        raise error_class(f"{cls._get_entity_type().capitalize()} with ID {entity_identifier} not found in the database.")

    return cls(entity_info, backend)

reload_data()

Reload the latest data for this entity from the database.

Raises:

Type Description
EntityNotFoundError

If an entry for this entity was not found in the database.
Source code in merlin/db_scripts/entities/db_entity.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
def reload_data(self):
    """
    Reload the latest data for this entity from the database.

    Raises:
        (exceptions.EntityNotFoundError): If an entry for this entity was not found in the database.
    """
    entity_id = self.get_id()
    updated_entity_info = self.backend.retrieve(entity_id, self.entity_type)
    if not updated_entity_info:
        error_class = self._error_classes.get(self.entity_type, EntityNotFoundError)
        raise error_class(f"{self.entity_type.capitalize()} with ID {entity_id} not found in the database.")
    self.entity_info = updated_entity_info

save()

Save the current state of this entity to the database.

Source code in merlin/db_scripts/entities/db_entity.py 
172
173
174
175
176
177
def save(self):
    """
    Save the current state of this entity to the database.
    """
    self.backend.save(self.entity_info)
    self._post_save_hook()