Skip to content

utils

Utility functions for backends in the Merlin application.

These utilities are essential for converting in-memory data models into a persistable format, ensuring compatibility with backend storage systems, and for consistent error handling when entity lookups fail.

deserialize_entity(data, model_class)

Given data that was retrieved, convert it into a data_class instance.

Parameters:

Name Type Description Default
data Dict[str, str]

The data retrieved that we need to deserialize.

 required
model_class T

A BaseDataModel subclass.

 required

Returns:

Type Description
T

A BaseDataModel instance.
Source code in merlin/backends/utils.py 
 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
def deserialize_entity(data: Dict[str, str], model_class: T) -> T:
    """
    Given data that was retrieved, convert it into a data_class instance.

    Args:
        data: The data retrieved that we need to deserialize.
        model_class: A [`BaseDataModel`][db_scripts.data_models.BaseDataModel] subclass.

    Returns:
        A [`BaseDataModel`][db_scripts.data_models.BaseDataModel] instance.
    """
    LOG.debug("Deserializing data...")
    deserialized_data = {}

    for key, val in data.items():
        if val.startswith("[") or val.startswith("{"):
            try:
                loaded_val = json.loads(val)
                # Check if this is a set that we specially encoded
                if isinstance(loaded_val, dict) and "__set__" in loaded_val:
                    deserialized_data[key] = set(loaded_val["__set__"])
                else:
                    deserialized_data[key] = loaded_val
            except json.JSONDecodeError as e:
                LOG.error(f"Failed to deserialize JSON for key {key}: {val}")
                LOG.error(f"Error: {str(e)}")
                # Use the original string value as fallback
                deserialized_data[key] = val
        elif val == "null":
            deserialized_data[key] = None
        elif val in ("True", "False"):
            deserialized_data[key] = val == "True"
        elif is_iso_datetime(val):
            deserialized_data[key] = datetime.fromisoformat(val)
        elif val.isdigit():
            deserialized_data[key] = float(val)
        else:
            deserialized_data[key] = str(val)

    LOG.debug("Successfully deserialized data.")
    return model_class.from_dict(deserialized_data)

get_not_found_error_class(model_class)

Get the appropriate not found error class based on the model type.

Parameters:

Name Type Description Default
model_class Type[T]

A BaseDataModel subclass.

 required

Returns:

Type Description
Exception

The error class to use.
Source code in merlin/backends/utils.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def get_not_found_error_class(model_class: Type[T]) -> Exception:
    """
    Get the appropriate not found error class based on the model type.

    Args:
        model_class: A [`BaseDataModel`][db_scripts.data_models.BaseDataModel] subclass.

    Returns:
        The error class to use.
    """
    error_map = {
        LogicalWorkerModel: WorkerNotFoundError,
        PhysicalWorkerModel: WorkerNotFoundError,
        RunModel: RunNotFoundError,
        StudyModel: StudyNotFoundError,
    }
    return error_map.get(model_class, Exception)

is_iso_datetime(value)

Check if a string is in ISO 8601 datetime format.

Parameters:

Name Type Description Default
value str

The string to check.

 required

Returns:

Type Description
bool

True if the string is in ISO 8601 format, False otherwise.
Source code in merlin/backends/utils.py 
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def is_iso_datetime(value: str) -> bool:
    """
    Check if a string is in ISO 8601 datetime format.

    Args:
        value: The string to check.

    Returns:
        True if the string is in ISO 8601 format, False otherwise.
    """
    try:
        datetime.fromisoformat(value)
        return True
    except ValueError:
        return False

serialize_entity(entity)

Given a BaseDataModel instance, convert it's data into a format that the database can interpret.

Parameters:

Name Type Description Default
entity T

A BaseDataModel instance.

 required

Returns:

Type Description
Dict[str, str]

A dictionary of information that the database can interpret.
Source code in merlin/backends/utils.py 
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
def serialize_entity(entity: T) -> Dict[str, str]:
    """
    Given a [`BaseDataModel`][db_scripts.data_models.BaseDataModel] instance,
    convert it's data into a format that the database can interpret.

    Args:
        entity: A [`BaseDataModel`][db_scripts.data_models.BaseDataModel] instance.

    Returns:
        A dictionary of information that the database can interpret.
    """
    LOG.debug("Serializing data...")
    serialized_data = {}

    for field in entity.get_instance_fields():
        field_value = getattr(entity, field.name)
        if isinstance(field_value, set):
            # Explicitly mark this as a set so we can properly deserialize it later
            serialized_data[field.name] = json.dumps({"__set__": list(field_value)})
        elif isinstance(field_value, (list, dict)):
            serialized_data[field.name] = json.dumps(field_value)
        elif isinstance(field_value, datetime):
            serialized_data[field.name] = field_value.isoformat()
        elif field_value is None:
            serialized_data[field.name] = "null"
        else:
            serialized_data[field.name] = str(field_value)

    LOG.debug("Successfully serialized data.")
    return serialized_data