Skip to content

BlueError

Bases: StackContextMixin, Exception

A unified, serializable exception class for the Blue platform.

BlueError is designed to wrap existing exceptions or create new ones with rich context, including:

  • Intent: How the system should react (e.g., 'fatal', 'retry').
  • Context: Arbitrary dictionary data attached to the error.
  • Logs: A timeline of messages added to the error as it bubbles up.
  • Stack Trace: A structured, serializable stack trace.

It supports "chaining" by initializing a new BlueError from an existing one, preserving the original timestamp and history while allowing new logs to be added.

Attributes:

Name Type Description
timestamp int

Unix timestamp (in milliseconds) of creation.

type str

The type name of the original exception (e.g., 'ValueError').

intent str

The severity or intended reaction (e.g., 'fatal', 'warning').

context Dict

Key-value pairs of metadata associated with the error.

log List[Dict]

A chronological list of log entries describing the error's journey.

stack_trace List[Dict]

structured stack trace frames.

Source code in blue/blueerror.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
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
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
class BlueError(StackContextMixin, Exception):
    """
    A unified, serializable exception class for the Blue platform.

    `BlueError` is designed to wrap existing exceptions or create new ones with
    rich context, including:

    - **Intent**: How the system should react (e.g., 'fatal', 'retry').
    - **Context**: Arbitrary dictionary data attached to the error.
    - **Logs**: A timeline of messages added to the error as it bubbles up.
    - **Stack Trace**: A structured, serializable stack trace.

    It supports "chaining" by initializing a new `BlueError` from an existing one,
    preserving the original timestamp and history while allowing new logs to be added.

    Attributes:
        timestamp (int): Unix timestamp (in milliseconds) of creation.
        type (str): The type name of the original exception (e.g., 'ValueError').
        intent (str): The severity or intended reaction (e.g., 'fatal', 'warning').
        context (Dict): Key-value pairs of metadata associated with the error.
        log (List[Dict]): A chronological list of log entries describing the error's journey.
        stack_trace (List[Dict]): structured stack trace frames.
    """

    def __init__(self, exception=None, description=None, intent="fatal", context={}):
        """
        Initialize the BlueError.

        Parameters:
            exception (Optional[Union[Exception, BlueError]]): The original exception to wrap.
                If it is already a `BlueError`, this acts as a copy/update constructor.
            description (Optional[str]): A human-readable description or log message
                to add immediately upon creation.
            intent (str, optional): The severity level or handling intent. Defaults to "fatal".
            context (Dict, optional): Additional metadata to merge into the error context.
        """
        super().__init__()
        if isinstance(exception, BlueError):
            existing_error = exception
            self.timestamp = existing_error.timestamp
            self.type = existing_error.type
            self.intent = existing_error.intent
            self.context = existing_error.context.copy() if existing_error.context else {}
            self.log = existing_error.log[:]
            self.stack_trace = existing_error.stack_trace[:]
            if description is not None:
                self.add_log(description, caller=get_instantiating_class_name())
            if context:
                pydash.objects.merge(self.context, context)
            return
        self.timestamp = int(time.time() * 1000)
        if exception:
            ex_type = type(exception)
            module_name = ex_type.__module__
            class_name = ex_type.__name__
            if module_name == 'builtins':
                self.type = class_name
            else:
                self.type = f"{module_name}.{class_name}"
        else:
            self.type = "Unknown BlueError"
        self.intent = intent
        self.context = context.copy() if context else {}
        self.log = []
        if description is not None:
            self.add_log(description, caller=get_instantiating_class_name())
        elif exception is not None:
            self.add_log(str(exception), caller=get_instantiating_class_name())
        exc_type, exc_value, exc_traceback = sys.exc_info()
        structured_trace = []
        if exc_traceback:
            frames = traceback.extract_tb(exc_traceback)
            for frame in frames:
                file_name = frame.filename.split('/')[-1]
                if not file_name == 'errorloom.py':
                    structured_trace.append(
                        {
                            'file': frame.filename,
                            'line_number': frame.lineno,
                            'function': frame.name,
                            'source_code': frame.line,
                        }
                    )
        self.stack_trace = structured_trace

    def set_intent(self, intent):
        """
        Updates the intent of the error (e.g., changing from 'fatal' to 'retry').

        Parameters:
            intent (str): The new intent string.
        """
        self.intent = intent

    def set_context(self, key, value):
        """
        Adds or updates a value in the error's context dictionary.
        Supports nested keys via dot notation (e.g., 'user.id').

        Parameters:
            key (str): The context key.
            value (Any): The value to store.
        """
        pydash.objects.set_(self.context, key, value)

    def add_log(self, description, caller=None):
        """
        Appends a new log entry to the error's history.

        Parameters:
            description (str): The message to log.
            caller (str, optional): The name of the calling class/component.
                If None, attempts to auto-detect the caller from the stack.
        """
        caller_name = self.get_calling_class_name(stack_depth=2)
        self.log.append({"timestamp": int(time.time() * 1000), "description": description, "caller": caller_name if caller is None else caller})

    def __str__(self):
        """
        Returns the JSON string representation of the error.
        """
        return json.dumps(self.get_dict())

    def get_dict(self):
        """
        Converts the BlueError into a serializable dictionary.

        Returns:
            Dict: A dictionary containing timestamp, type, intent, log, and stack_trace.
        """
        return {"timestamp": self.timestamp, "type": self.type, "intent": self.intent, "log": self.log, "stack_trace": self.stack_trace}

    def from_json(self, json_string):
        """
        Hydrates this BlueError instance from a JSON string.

        Parameters:
            json_string (str): The JSON representation of a BlueError.
        """
        data = json.loads(json_string)
        self.timestamp = pydash.objects.get(data, 'timestamp', int(time.time() * 1000))
        self.type = pydash.objects.get(data, 'type', 'Unknown BlueError')
        self.intent = pydash.objects.get(data, 'intent', 'fatal')
        self.log = pydash.objects.get(data, 'log', [])
        self.stack_trace = pydash.objects.get(data, 'stack_trace', [])

__init__(exception=None, description=None, intent='fatal', context={})

Initialize the BlueError.

Parameters:

Name Type Description Default
exception Optional[Union[Exception, BlueError]]

The original exception to wrap. If it is already a BlueError, this acts as a copy/update constructor.

None
description Optional[str]

A human-readable description or log message to add immediately upon creation.

None
intent str

The severity level or handling intent. Defaults to "fatal".

'fatal'
context Dict

Additional metadata to merge into the error context.

{}
Source code in blue/blueerror.py
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
def __init__(self, exception=None, description=None, intent="fatal", context={}):
    """
    Initialize the BlueError.

    Parameters:
        exception (Optional[Union[Exception, BlueError]]): The original exception to wrap.
            If it is already a `BlueError`, this acts as a copy/update constructor.
        description (Optional[str]): A human-readable description or log message
            to add immediately upon creation.
        intent (str, optional): The severity level or handling intent. Defaults to "fatal".
        context (Dict, optional): Additional metadata to merge into the error context.
    """
    super().__init__()
    if isinstance(exception, BlueError):
        existing_error = exception
        self.timestamp = existing_error.timestamp
        self.type = existing_error.type
        self.intent = existing_error.intent
        self.context = existing_error.context.copy() if existing_error.context else {}
        self.log = existing_error.log[:]
        self.stack_trace = existing_error.stack_trace[:]
        if description is not None:
            self.add_log(description, caller=get_instantiating_class_name())
        if context:
            pydash.objects.merge(self.context, context)
        return
    self.timestamp = int(time.time() * 1000)
    if exception:
        ex_type = type(exception)
        module_name = ex_type.__module__
        class_name = ex_type.__name__
        if module_name == 'builtins':
            self.type = class_name
        else:
            self.type = f"{module_name}.{class_name}"
    else:
        self.type = "Unknown BlueError"
    self.intent = intent
    self.context = context.copy() if context else {}
    self.log = []
    if description is not None:
        self.add_log(description, caller=get_instantiating_class_name())
    elif exception is not None:
        self.add_log(str(exception), caller=get_instantiating_class_name())
    exc_type, exc_value, exc_traceback = sys.exc_info()
    structured_trace = []
    if exc_traceback:
        frames = traceback.extract_tb(exc_traceback)
        for frame in frames:
            file_name = frame.filename.split('/')[-1]
            if not file_name == 'errorloom.py':
                structured_trace.append(
                    {
                        'file': frame.filename,
                        'line_number': frame.lineno,
                        'function': frame.name,
                        'source_code': frame.line,
                    }
                )
    self.stack_trace = structured_trace

__str__()

Returns the JSON string representation of the error.

Source code in blue/blueerror.py
214
215
216
217
218
def __str__(self):
    """
    Returns the JSON string representation of the error.
    """
    return json.dumps(self.get_dict())

add_log(description, caller=None)

Appends a new log entry to the error's history.

Parameters:

Name Type Description Default
description str

The message to log.

required
caller str

The name of the calling class/component. If None, attempts to auto-detect the caller from the stack.

None
Source code in blue/blueerror.py
202
203
204
205
206
207
208
209
210
211
212
def add_log(self, description, caller=None):
    """
    Appends a new log entry to the error's history.

    Parameters:
        description (str): The message to log.
        caller (str, optional): The name of the calling class/component.
            If None, attempts to auto-detect the caller from the stack.
    """
    caller_name = self.get_calling_class_name(stack_depth=2)
    self.log.append({"timestamp": int(time.time() * 1000), "description": description, "caller": caller_name if caller is None else caller})

from_json(json_string)

Hydrates this BlueError instance from a JSON string.

Parameters:

Name Type Description Default
json_string str

The JSON representation of a BlueError.

required
Source code in blue/blueerror.py
229
230
231
232
233
234
235
236
237
238
239
240
241
def from_json(self, json_string):
    """
    Hydrates this BlueError instance from a JSON string.

    Parameters:
        json_string (str): The JSON representation of a BlueError.
    """
    data = json.loads(json_string)
    self.timestamp = pydash.objects.get(data, 'timestamp', int(time.time() * 1000))
    self.type = pydash.objects.get(data, 'type', 'Unknown BlueError')
    self.intent = pydash.objects.get(data, 'intent', 'fatal')
    self.log = pydash.objects.get(data, 'log', [])
    self.stack_trace = pydash.objects.get(data, 'stack_trace', [])

get_dict()

Converts the BlueError into a serializable dictionary.

Returns:

Name Type Description
Dict

A dictionary containing timestamp, type, intent, log, and stack_trace.

Source code in blue/blueerror.py
220
221
222
223
224
225
226
227
def get_dict(self):
    """
    Converts the BlueError into a serializable dictionary.

    Returns:
        Dict: A dictionary containing timestamp, type, intent, log, and stack_trace.
    """
    return {"timestamp": self.timestamp, "type": self.type, "intent": self.intent, "log": self.log, "stack_trace": self.stack_trace}

set_context(key, value)

Adds or updates a value in the error's context dictionary. Supports nested keys via dot notation (e.g., 'user.id').

Parameters:

Name Type Description Default
key str

The context key.

required
value Any

The value to store.

required
Source code in blue/blueerror.py
191
192
193
194
195
196
197
198
199
200
def set_context(self, key, value):
    """
    Adds or updates a value in the error's context dictionary.
    Supports nested keys via dot notation (e.g., 'user.id').

    Parameters:
        key (str): The context key.
        value (Any): The value to store.
    """
    pydash.objects.set_(self.context, key, value)

set_intent(intent)

Updates the intent of the error (e.g., changing from 'fatal' to 'retry').

Parameters:

Name Type Description Default
intent str

The new intent string.

required
Source code in blue/blueerror.py
182
183
184
185
186
187
188
189
def set_intent(self, intent):
    """
    Updates the intent of the error (e.g., changing from 'fatal' to 'retry').

    Parameters:
        intent (str): The new intent string.
    """
    self.intent = intent
Last update: 2026-02-04