Guides
Advanced error handling
Build a structured error taxonomy with typed error codes, user-facing messages, and automatic exception classification.
In this guide you will replace generic exception handling with a typed error system that distinguishes who caused the error, what the user can do about it, and how the system should respond.
Prerequisites
- Familiarity with Basic error handling
Why structured errors
A raw exception like ValueError("plate not found") gives you a string. A structured error gives you:
- Who is responsible: user mistake, system failure, config problem, or application bug
- What the user sees: a clear title, explanation, and actionable steps
- What operations can do: an error code to look up, and whether a retry is safe
- What developers see: full technical details in the logs
Define the error types
Start with an enum that classifies every error by who caused it:
from enum import Enum
class ErrorType(Enum):
USER_ERROR = "user_error" # Bad input or invalid selection — user can fix
SYSTEM_ERROR = "system_error" # Transient failure — retry might help
CONFIG_ERROR = "config_error" # Missing credentials or misconfiguration — admin must fix
BUG = "bug" # Unexpected application error — developer must fix
The StructuredError dataclass
A StructuredError carries everything needed to handle an error consistently at every layer of the system:
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class StructuredError:
error_type: ErrorType
title: str # Short, human-readable title
message: str # Explanation of what went wrong
actions: list[str] = field(default_factory=list) # Steps the user can take
error_code: Optional[str] = None # Unique code for lookup (e.g. "WF_101")
can_retry: bool = False # Whether re-running the workflow may succeed
contact_support: bool = False # Whether to escalate to support
technical_details: Optional[str] = None # Stack trace / raw exception (omit from UI)
def to_user_message(self) -> str:
"""Format the error for display to an end user."""
lines = [f"**{self.title}**", "", self.message]
if self.actions:
lines += ["", "**What you can do:**"]
lines += [f"- {action}" for action in self.actions]
if self.error_code:
lines += ["", f"Error code: `{self.error_code}`"]
if self.contact_support:
lines += ["", "If the problem persists, contact support."]
return "\n".join(lines)
Factory functions
Define one factory function per error type so callers never need to import ErrorType directly:
def create_user_error(
title: str,
message: str,
actions: list[str] | None = None,
error_code: str | None = None,
can_retry: bool = False,
) -> StructuredError:
return StructuredError(
error_type=ErrorType.USER_ERROR,
title=title,
message=message,
actions=actions or [],
error_code=error_code,
can_retry=can_retry,
)
def create_system_error(
title: str,
message: str,
actions: list[str] | None = None,
error_code: str | None = None,
can_retry: bool = True, # System errors are often transient
technical_details: str | None = None,
) -> StructuredError:
return StructuredError(
error_type=ErrorType.SYSTEM_ERROR,
title=title,
message=message,
actions=actions or ["Wait a moment and try again", "Contact support if the problem persists"],
error_code=error_code,
can_retry=can_retry,
contact_support=True,
technical_details=technical_details,
)
def create_config_error(
title: str,
message: str,
actions: list[str] | None = None,
error_code: str | None = None,
) -> StructuredError:
return StructuredError(
error_type=ErrorType.CONFIG_ERROR,
title=title,
message=message,
actions=actions or ["Contact your administrator"],
error_code=error_code,
can_retry=False,
contact_support=True,
)
Error codes
Assign numeric codes to every distinct error so operators can look them up in your runbook without reading a full stack trace.
A simple numbering convention:
| Range | Category |
|---|---|
WF_100 – WF_199 | User input errors |
WF_200 – WF_299 | Data validation errors |
WF_300 – WF_399 | External API errors |
WF_400 – WF_499 | Configuration errors |
WF_500 – WF_599 | Instrument / hardware errors |
class ErrorCode:
# User input
NO_SAMPLES_SELECTED = "WF_100"
INVALID_SAMPLE_ID = "WF_101"
INVALID_PARAMETER_VALUE = "WF_102"
# Data
NO_DATA_FOUND = "WF_200"
MISSING_INSTRUMENT_DATA = "WF_201"
# External APIs
API_TIMEOUT = "WF_300"
API_AUTH_FAILED = "WF_301"
API_NOT_FOUND = "WF_302"
# Configuration
MISSING_CREDENTIALS = "WF_400"
MISSING_ENV_VAR = "WF_401"
# Instruments
INSTRUMENT_UNREACHABLE = "WF_500"
INSTRUMENT_RUN_FAILED = "WF_501"
Pre-built error constructors
For each domain-specific error condition, write a named constructor. This keeps error definitions in one place and makes intent obvious at the call site.
def no_samples_error() -> StructuredError:
return create_user_error(
title="No Samples Selected",
message="At least one sample must be selected to start the workflow.",
actions=[
"Select one or more samples from the list",
"Verify the samples exist in the system",
],
error_code=ErrorCode.NO_SAMPLES_SELECTED,
can_retry=True,
)
def instrument_unreachable_error(instrument_name: str) -> StructuredError:
return create_system_error(
title=f"{instrument_name} Unreachable",
message=f"Could not connect to {instrument_name}. The instrument may be offline or the connector may have stopped.",
actions=[
f"Check that {instrument_name} is powered on",
"Verify the connector service is running",
"Restart the connector if needed",
],
error_code=ErrorCode.INSTRUMENT_UNREACHABLE,
can_retry=True,
)
def missing_env_var_error(var_name: str) -> StructuredError:
return create_config_error(
title="Missing Configuration",
message=f"Required environment variable '{var_name}' is not set.",
actions=[
f"Set {var_name} in your .env file or CI/CD secrets",
"Contact your administrator if you do not have access",
],
error_code=ErrorCode.MISSING_ENV_VAR,
)
Raise structured errors from a flow
Wrap StructuredError in an exception class so it can be raised and caught naturally:
class WorkflowError(Exception):
def __init__(self, structured_error: StructuredError) -> None:
super().__init__(structured_error.title)
self.structured_error = structured_error
# Raising:
sample_ids = []
if not sample_ids:
raise WorkflowError(no_samples_error())
Handle it in the flow:
from unitelabs.sdk import get_logger
from unitelabs.sdk.automate import workflow
@workflow(log_prints=True, name="Structured Error Example")
async def process_flow(sample_ids: list[str]) -> dict:
logger = get_logger()
try:
if not sample_ids:
raise WorkflowError(no_samples_error())
# ... actual processing
return {"status": "success"}
except WorkflowError as e:
err = e.structured_error
# Log full details for developers
logger.error(
f"[{err.error_code}] {err.error_type.value}: {err.title}\n"
f"Message: {err.message}\n"
f"Can retry: {err.can_retry}"
)
# Return structured info for callers
return {
"status": "error",
"error_type": err.error_type.value,
"error_code": err.error_code,
"title": err.title,
"message": err.message,
"actions": err.actions,
"can_retry": err.can_retry,
}
Automatic exception classification
Add a classify_exception function to map common Python exceptions to the right StructuredError type. This means you catch and classify Exception once in the workflow rather than writing specific handlers for every exception type.
import httpx
def classify_exception(exc: Exception, context: str = "") -> StructuredError:
"""Map a raw exception to a StructuredError based on its type."""
# HTTP errors from API calls
if isinstance(exc, httpx.HTTPStatusError):
status = exc.response.status_code
if status in (401, 403):
return create_config_error(
title="Authentication Failed",
message="API credentials are invalid or expired.",
actions=["Contact your administrator to refresh the API credentials"],
error_code=ErrorCode.API_AUTH_FAILED,
)
if status == 404:
return create_user_error(
title="Resource Not Found",
message=f"The requested resource does not exist{' while ' + context if context else ''}.",
actions=["Verify the ID is correct", "Check that the resource has not been deleted"],
error_code=ErrorCode.API_NOT_FOUND,
)
if status >= 500:
return create_system_error(
title="API Server Error",
message="The API returned an unexpected server error.",
error_code=ErrorCode.API_TIMEOUT,
technical_details=str(exc),
)
if isinstance(exc, httpx.TimeoutException):
return create_system_error(
title="Request Timed Out",
message="The API did not respond in time.",
error_code=ErrorCode.API_TIMEOUT,
can_retry=True,
)
# Missing environment variables
if isinstance(exc, (KeyError, RuntimeError)) and "not set" in str(exc).lower():
return create_config_error(
title="Missing Configuration",
message=str(exc),
error_code=ErrorCode.MISSING_ENV_VAR,
)
# Bad user input
if isinstance(exc, ValueError):
return create_user_error(
title="Invalid Input",
message=str(exc),
actions=["Review your input and try again"],
)
# Fallback — unexpected bug
return StructuredError(
error_type=ErrorType.BUG,
title="Unexpected Error",
message=f"An unexpected error occurred{' while ' + context if context else ''}.",
contact_support=True,
can_retry=False,
technical_details=str(exc),
)
Use it in the flow's catch-all:
@workflow(log_prints=True, name="Classified Error Flow")
async def classified_flow(source: str) -> dict:
logger = get_logger()
try:
data = await fetch_data_task(source=source)
return {"status": "success", "count": len(data)}
except WorkflowError as e:
# Already structured — use as-is
err = e.structured_error
except Exception as e:
# Classify automatically
err = classify_exception(e, context="fetching data")
logger.error(f"[{err.error_code}] {err.title}: {err.message}")
return {
"status": "error",
"error_type": err.error_type.value,
"error_code": err.error_code,
"title": err.title,
"message": err.message,
"actions": err.actions,
"can_retry": err.can_retry,
}
What users see vs. what developers see
The same StructuredError produces different outputs depending on context:
User-facing message (e.g. displayed in a UI or returned to a caller):
**No Samples Selected**
At least one sample must be selected to start the workflow.
**What you can do:**
- Select one or more samples from the list
- Verify the samples exist in the system
Error code: `WF_100`
Developer log (Prefect run logs):
[WF_100] user_error: No Samples Selected
Message: At least one sample must be selected to start the workflow.
Can retry: True
Keep technical_details (stack traces, raw API responses) out of user-facing messages. Pass them only to logs or an internal support system.