`TaskException` model reference¶

TaskException is the JSON-serializable record of a task or workflow failure: the exception's type, module, message, sanitized args, and formatted traceback. It is stored on task.exception (and workflow.exception) when execution fails, so failures can be inspected from any process. to_exception() rebuilds the original exception on a best-effort basis, falling back to TaskFailedError when the class cannot be imported or constructed.

sheppy.models.TaskException ¶

Bases: BaseModel

Serializable representation of an exception raised during task execution.

Captures everything needed to inspect a task failure in a JSON-serializable form, so it can be stored in any backend and reconstructed later.

ATTRIBUTE	DESCRIPTION
`type`	Exception class name, e.g. `ValueError`. TYPE: `str`
`module`	Module where the exception class is defined, e.g. `builtins`. TYPE: `str`
`message`	Exception message, i.e. `str(exception)`. TYPE: `str`
`args`	Exception args, sanitized to be JSON serializable (non-serializable values are replaced with their repr). TYPE: `tuple[Any, ...]`
`traceback`	Full formatted traceback of the exception. None for synthesized exceptions (e.g. worker crash). TYPE: `str \| None`

Example

from sheppy import Queue, task

@task
def failing_task():
    raise ValueError("something went wrong")

q = Queue(...)
t = await q.add(failing_task())
assert t is not None
t = await q.wait_for(t)

assert t.exception is not None
print(t.exception.type)       # "ValueError"
print(t.exception.message)    # "something went wrong"
print(t.exception.traceback)  # full formatted traceback

# best-effort reconstruction of the original exception
exc = t.exception.to_exception()
assert isinstance(exc, ValueError)

type ¶

type: str

str: Exception class name, e.g. ValueError.

module ¶

module: str

str: Module where the exception class is defined, e.g. builtins.

message ¶

message: str

str: Exception message, i.e. str(exception).

args ¶

args: tuple[Any, ...] = Field(default_factory=tuple)

tuple[Any, ...]: Exception args, sanitized to be JSON serializable.

traceback ¶

traceback: str | None = None

str|None: Full formatted traceback. None for synthesized exceptions (e.g. worker crash).

from_exception ¶

from_exception(exception: BaseException) -> TaskException

Source code in src/sheppy/models.py

@classmethod
def from_exception(cls, exception: BaseException) -> 'TaskException':
    return cls(
        type=type(exception).__qualname__,
        module=type(exception).__module__,
        message=str(exception),
        args=tuple(_json_safe(arg) for arg in exception.args),
        traceback="".join(format_exception(exception)),
    )

to_exception ¶

to_exception() -> Exception