Skip to content

run_management

Module for managing entities and their associated runs in the database.

This module provides a mixin class, RunManagementMixin, designed to simplify the management of runs associated with an entity. The mixin can be used by any class that has the necessary attributes and methods to support run management, such as reload_data, save, and an entity_info object containing a runs list.

RunManagementMixin

Mixin for managing runs associated with an entity.

This mixin provides utility methods for handling run IDs associated with an entity. It assumes that the class using this mixin has the necessary attributes and methods to support run management, including reload_data, save, and an entity_info object containing a runs list.

Methods:

Name Description
get_runs

Retrieve the IDs of the runs associated with the entity.
add_run

Add a run ID to the list of runs.
remove_run

Remove a run ID from the list of runs.
Source code in merlin/db_scripts/entities/mixins/run_management.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 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
class RunManagementMixin:
    """
    Mixin for managing runs associated with an entity.

    This mixin provides utility methods for handling run IDs associated with an entity.
    It assumes that the class using this mixin has the necessary attributes and methods
    to support run management, including `reload_data`, `save`, and an `entity_info` object
    containing a `runs` list.

    Methods:
        get_runs:
            Retrieve the IDs of the runs associated with the entity.

        add_run:
            Add a run ID to the list of runs.

        remove_run:
            Remove a run ID from the list of runs.
    """

    def get_runs(self) -> List[str]:
        """
        Get every run listed in this entity.

        Assumptions:
            - The class using this must have a `reload_data` method
            - The class using this must have an `entity_info` object containing a `runs` list

        Returns:
            A list of run IDs.
        """
        self.reload_data()
        return self.entity_info.runs

    def add_run(self, run_id: str):
        """
        Add a new run ID to the list of runs.

        Assumptions:
            - The class using this must have an `entity_info` object containing a `runs` list
            - The class using this must have a `save` method

        Args:
            run_id: The ID of the run to add.
        """
        self.entity_info.runs.append(run_id)
        self.save()

    def remove_run(self, run_id: str):
        """
        Remove a run ID from the list of runs.

        Does *not* delete the run entity from the database. This will only remove the run's ID
        from the list in this entity.

        Assumptions:
            - The class using this must have a `reload_data` method
            - The class using this must have an `entity_info` object containing a `runs` list
            - The class using this must have a `save` method

        Args:
            run_id: The ID of the run to remove.
        """
        self.reload_data()
        self.entity_info.runs.remove(run_id)
        self.save()

    def construct_run_string(self) -> str:
        """
        Constructs and returns a formatted string representation of all runs associated
        with the current instance.

        Returns:
            A formatted string containing details of all runs.
        """
        from merlin.db_scripts.entities.run_entity import RunEntity  # pylint: disable=import-outside-toplevel

        runs = self.get_runs()
        run_str = ""
        if runs:
            for run_id in runs:
                try:
                    run = RunEntity.load(run_id, self.backend)
                    run_str += f"  - ID: {run.get_id()}\n    Workspace: {run.get_workspace()}\n"
                except Exception:  # pylint: disable=broad-exception-caught
                    run_str += f"  - ID: {run_id} (Error loading run)\n"
        else:
            run_str = "  No runs found.\n"
        return run_str

add_run(run_id)

Add a new run ID to the list of runs.

Assumptions
  • The class using this must have an entity_info object containing a runs list
  • The class using this must have a save method

Parameters:

Name Type Description Default
run_id str

The ID of the run to add.

 required
Source code in merlin/db_scripts/entities/mixins/run_management.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
def add_run(self, run_id: str):
    """
    Add a new run ID to the list of runs.

    Assumptions:
        - The class using this must have an `entity_info` object containing a `runs` list
        - The class using this must have a `save` method

    Args:
        run_id: The ID of the run to add.
    """
    self.entity_info.runs.append(run_id)
    self.save()

construct_run_string()

Constructs and returns a formatted string representation of all runs associated with the current instance.

Returns:

Type Description
str

A formatted string containing details of all runs.
Source code in merlin/db_scripts/entities/mixins/run_management.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def construct_run_string(self) -> str:
    """
    Constructs and returns a formatted string representation of all runs associated
    with the current instance.

    Returns:
        A formatted string containing details of all runs.
    """
    from merlin.db_scripts.entities.run_entity import RunEntity  # pylint: disable=import-outside-toplevel

    runs = self.get_runs()
    run_str = ""
    if runs:
        for run_id in runs:
            try:
                run = RunEntity.load(run_id, self.backend)
                run_str += f"  - ID: {run.get_id()}\n    Workspace: {run.get_workspace()}\n"
            except Exception:  # pylint: disable=broad-exception-caught
                run_str += f"  - ID: {run_id} (Error loading run)\n"
    else:
        run_str = "  No runs found.\n"
    return run_str

get_runs()

Get every run listed in this entity.

Assumptions
  • The class using this must have a reload_data method
  • The class using this must have an entity_info object containing a runs list

Returns:

Type Description
List[str]

A list of run IDs.
Source code in merlin/db_scripts/entities/mixins/run_management.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
def get_runs(self) -> List[str]:
    """
    Get every run listed in this entity.

    Assumptions:
        - The class using this must have a `reload_data` method
        - The class using this must have an `entity_info` object containing a `runs` list

    Returns:
        A list of run IDs.
    """
    self.reload_data()
    return self.entity_info.runs

remove_run(run_id)

Remove a run ID from the list of runs.

Does not delete the run entity from the database. This will only remove the run's ID from the list in this entity.

Assumptions
  • The class using this must have a reload_data method
  • The class using this must have an entity_info object containing a runs list
  • The class using this must have a save method

Parameters:

Name Type Description Default
run_id str

The ID of the run to remove.

 required
Source code in merlin/db_scripts/entities/mixins/run_management.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def remove_run(self, run_id: str):
    """
    Remove a run ID from the list of runs.

    Does *not* delete the run entity from the database. This will only remove the run's ID
    from the list in this entity.

    Assumptions:
        - The class using this must have a `reload_data` method
        - The class using this must have an `entity_info` object containing a `runs` list
        - The class using this must have a `save` method

    Args:
        run_id: The ID of the run to remove.
    """
    self.reload_data()
    self.entity_info.runs.remove(run_id)
    self.save()