Source code for fennel.exceptions

from typing import List

[docs]class FennelException(Exception): pass
[docs]class TaskFailed(FennelException): def __init__(self, original_type: str, original_args: List): """ This exception is returned by worker processes which experienced an exception when executing a task. Parameters ---------- original_type : str The name of the original exception, e.g. ``'ValueError'``. original_args : List The arguments given to the original exception, e.g. ``['Not found']`` Examples -------- >>> @app.task(retries=0) >>> async def foo(n): ... raise Exception("baz") ... >>> x = await foo.delay(3) >>> try: ... result = await x.get() >>> except TaskFailed as e: ... assert e.original_type == "Exception" ... assert e.original_args == ["baz"] """ self.original_type = original_type self.original_args = original_args
[docs]class ResultsDisabled(FennelException): """ Raised when ``results_enabled=False`` and code attempts to access a tasks result via ``.get()``. """
[docs]class UnknownTask(FennelException): """ Raised by a worker process if it is unable to find a Python function corresponding to the task it has read from the queue. """
[docs]class Timeout(FennelException): """ Raised by client code when a given timeout is exceeded when waiting for results to arrive. """
[docs]class JobNotFound(FennelException): """ Raised by client code when attempting to retrieve job information that cannot be found in Redis. """
[docs]class Chaos(FennelException): """ Used in tests to ensure failures are handled properly. """
[docs]class Completed(FennelException): """ Used internally to shutdown an Executor if the exit condition is completing all tasks. """