taskiq/docs/examples/errors/error_middleware.py at 56512dd7152e2b2dd354d454d8ad28aaeaffa79b · taskiq-python/taskiq · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from typing import Any

from taskiq import InMemoryBroker, TaskiqMessage, TaskiqMiddleware, TaskiqResult


class ErrorReporterMiddleware(TaskiqMiddleware):
    def on_error(
        self,
        message: "TaskiqMessage",
        result: "TaskiqResult[Any]",
        exception: BaseException,
    ) -> None:
        # This code runs on the worker, so the traceback
        # of the exception is still available here.
        print(
            f"Task {message.task_name} with id {message.task_id} "
            f"failed: {exception!r}",
        )


broker = InMemoryBroker().with_middlewares(ErrorReporterMiddleware())