Skip to content

Commit 56512dd

Browse files
committed
docs: clarify part about error handling in docs
1 parent 8a1b04d commit 56512dd

8 files changed

Lines changed: 211 additions & 9 deletions

File tree

‎docs/contrib.md‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,12 +26,12 @@ uv sync --all-extras
2626

2727
## Linting
2828

29-
We have `pre-commit` configured with all our settings. We highly recommend you to install it as a git hook using `pre-commit install` command.
29+
We have `prek` configured with all our settings. We highly recommend you to install it as a git hook using `prek install` command.
3030

3131
But even without installation, you can run all lints manually:
3232

3333
```bash
34-
pre-commit run -a
34+
prek run -a
3535
```
3636

3737
## Testing
Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
from typing import Any
2+
3+
from taskiq import InMemoryBroker, TaskiqMessage, TaskiqMiddleware, TaskiqResult
4+
5+
6+
class ErrorReporterMiddleware(TaskiqMiddleware):
7+
def on_error(
8+
self,
9+
message: "TaskiqMessage",
10+
result: "TaskiqResult[Any]",
11+
exception: BaseException,
12+
) -> None:
13+
# This code runs on the worker, so the traceback
14+
# of the exception is still available here.
15+
print(
16+
f"Task {message.task_name} with id {message.task_id} "
17+
f"failed: {exception!r}",
18+
)
19+
20+
21+
broker = InMemoryBroker().with_middlewares(ErrorReporterMiddleware())
Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
import asyncio
2+
from typing import Any
3+
4+
from taskiq import InMemoryBroker
5+
6+
broker = InMemoryBroker()
7+
8+
9+
@broker.task
10+
async def divide(a: int, b: int) -> Any:
11+
return a / b
12+
13+
14+
async def main() -> None:
15+
await broker.startup()
16+
17+
task = await divide.kiq(1, 0)
18+
try:
19+
# Raises ZeroDivisionError, otherwise returns the result itself.
20+
result = (await task.wait_result()).raise_for_error()
21+
except ZeroDivisionError:
22+
print("Cannot divide by zero.")
23+
else:
24+
print(f"Returned value: {result.return_value}")
25+
26+
await broker.shutdown()
27+
28+
29+
if __name__ == "__main__":
30+
asyncio.run(main())
Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
import asyncio
2+
from typing import Any
3+
4+
from taskiq import InMemoryBroker
5+
6+
broker = InMemoryBroker()
7+
8+
9+
@broker.task
10+
async def divide(a: int, b: int) -> Any:
11+
return a / b
12+
13+
14+
async def main() -> None:
15+
await broker.startup()
16+
17+
task = await divide.kiq(1, 0)
18+
result = await task.wait_result()
19+
20+
if result.is_err:
21+
print(f"Task failed with {type(result.error).__name__}: {result.error}")
22+
else:
23+
print(f"Returned value: {result.return_value}")
24+
25+
await broker.shutdown()
26+
27+
28+
if __name__ == "__main__":
29+
asyncio.run(main())

‎docs/examples/state/events_example.py‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -45,10 +45,9 @@ async def main() -> None:
4545
await broker.startup()
4646

4747
set_task = await set_val.kiq("key", "value")
48-
set_result = await set_task.wait_result(with_logs=True)
48+
set_result = await set_task.wait_result()
4949
if set_result.is_err:
50-
print(set_result.log)
51-
raise ValueError("Cannot set value in redis. See logs.")
50+
raise ValueError("Cannot set value in redis.") from set_result.error
5251

5352
get_task = await get_val.kiq("key")
5453
get_res = await get_task.wait_result()

‎docs/examples/state/events_example_annot.py‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -53,10 +53,9 @@ async def main() -> None:
5353
await broker.startup()
5454

5555
set_task = await set_val.kiq("key", "value")
56-
set_result = await set_task.wait_result(with_logs=True)
56+
set_result = await set_task.wait_result()
5757
if set_result.is_err:
58-
print(set_result.log)
59-
raise ValueError("Cannot set value in redis. See logs.")
58+
raise ValueError("Cannot set value in redis.") from set_result.error
6059

6160
get_task = await get_val.kiq("key")
6261
get_res = await get_task.wait_result()

‎docs/guide/architecture-overview.md‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -137,6 +137,8 @@ async def my_async_task() -> None:
137137

138138
Result backend is used to store and get results of the execution.
139139
Results have type `TaskiqResult` from [taskiq.result](https://github.com/taskiq-python/taskiq/blob/master/taskiq/result.py).
140+
It holds the returned value, the execution time and the exception, if the task has failed.
141+
See [Error handling](./error-handling.md) for more info about failed tasks.
140142

141143
Every ResultBackend must implement `AsyncResultBackend` from [taskiq.abc.result_backend](https://github.com/taskiq-python/taskiq/blob/master/taskiq/abc/result_backend.py). By default, brokers use `DummyResultBackend`. It doesn't do anything and cannot be used
142144
in real-world scenarios. But some brokers can override it. For example `InMemoryBroker` by default uses `InMemoryResultBackend` and returns correct results.
@@ -223,7 +225,7 @@ Here are methods you can implement in the order they are executed:
223225
- `pre_send` - executed on the client side before the message is sent. Here you can modify the message.
224226
- `post_send` - executed right after the message was sent.
225227
- `pre_execute` - executed on the worker side after the message was received by a worker and before its execution.
226-
- `on_error` - executed after the task was executed if an exception was found.
228+
- `on_error` - executed after the task was executed if an exception was found. See [Error handling](./error-handling.md).
227229
- `post_execute` - executed after the message was executed.
228230
- `post_save` - executed after the result was saved in the result backend.
229231

‎docs/guide/error-handling.md‎

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,122 @@
1+
---
2+
title: Error handling
3+
order: 10
4+
---
5+
6+
# Error handling
7+
8+
When a task raises an exception, the worker doesn't crash. It catches the exception, logs it and stores it in the
9+
`TaskiqResult` along with everything else. What you do with it afterwards depends on where you want to handle the error:
10+
on the client side, where the result is awaited, or on the worker side, where the exception happened.
11+
12+
## Handling errors on the client
13+
14+
`TaskiqResult` has two fields related to errors:
15+
16+
* `is_err` - `True` if the task has failed;
17+
* `error` - the exception itself, or `None` if the task succeeded.
18+
19+
@[code python](../examples/errors/result_error.py)
20+
21+
If you'd rather work with exceptions than with flags, use the `raise_for_error` method. It raises the stored exception
22+
if there is one and returns the result itself otherwise, so it can be chained right after `wait_result`.
23+
24+
@[code python](../examples/errors/raise_for_error.py)
25+
26+
Tracebacks are not transferred to the client. Only the exception type, its arguments and the chain of `__cause__`/`__context__`
27+
exceptions are serialized. Sending tracebacks over the network is unreliable and may leak sensitive information from the
28+
worker, so Taskiq intentionally doesn't do it.
29+
30+
Also, the client has to be able to import the exception class to restore it. If the exception is defined in a module
31+
unavailable on the client side, Taskiq creates a synthetic exception class with the same name in the `taskiq.exceptions`
32+
namespace, so `except MyError` won't catch it, while `except Exception` still will.
33+
34+
## Handling errors on the worker
35+
36+
Client-side handling is only possible if you use a result backend and wait for the result. Fire-and-forget tasks have
37+
nobody to report to. Besides, sometimes you want to react to every failure in one place instead of duplicating
38+
the same handling in every caller.
39+
40+
For these cases, use the `on_error` hook of a middleware. It's called on the worker right after the exception is caught, so the full traceback is still available.
41+
42+
@[code python](../examples/errors/error_middleware.py)
43+
44+
That's the recommended way to integrate taskiq with error tracking systems. For example, here's how you can send all task failures to Sentry:
45+
46+
```python
47+
from typing import Any
48+
49+
import sentry_sdk
50+
from sentry_sdk.integrations.logging import LoggingIntegration, ignore_logger
51+
52+
from taskiq import (
53+
TaskiqEvents,
54+
TaskiqMessage,
55+
TaskiqMiddleware,
56+
TaskiqResult,
57+
TaskiqState,
58+
)
59+
60+
61+
class SentryMiddleware(TaskiqMiddleware):
62+
def on_error(
63+
self,
64+
message: "TaskiqMessage",
65+
result: "TaskiqResult[Any]",
66+
exception: BaseException,
67+
) -> None:
68+
sentry_sdk.capture_exception(
69+
exception,
70+
tags={
71+
"taskiq_task_name": message.task_name,
72+
"taskiq_task_id": message.task_id,
73+
},
74+
)
75+
76+
77+
# Here `broker` is the broker you defined for your project.
78+
broker = broker.with_middlewares(SentryMiddleware())
79+
80+
81+
@broker.on_event(TaskiqEvents.WORKER_STARTUP)
82+
def init_sentry(_: TaskiqState) -> None:
83+
# We ignore this logger, because its events may overwrite the exception
84+
# we capture in the middleware with new tags.
85+
ignore_logger("taskiq.receiver.receiver")
86+
sentry_sdk.init(
87+
"https://key@your-sentry-instance",
88+
environment="dev",
89+
integrations=[LoggingIntegration()],
90+
)
91+
```
92+
93+
If your framework integration initializes Sentry on its own, you can drop the startup event and keep the middleware only.
94+
95+
::: tip Retries
96+
97+
If you want failed tasks to be retried instead of being reported, take a look at the [`SimpleRetryMiddleware`](../available-components/middlewares.md)
98+
or the [`SmartRetryMiddleware`](../available-components/middlewares.md).
99+
100+
:::
101+
102+
## Errors in dependencies
103+
104+
Generator dependencies can catch exceptions raised by the task they are injected into. It's useful if you want to roll
105+
back a transaction that was opened in the dependency. This is described in
106+
[State and Dependencies](./state-and-deps.md#exception-handling), along with the `--no-propagate-errors` option
107+
that turns this behaviour off.
108+
109+
## Skipping the result
110+
111+
Sometimes a task shouldn't store any result at all. Raising `NoResultError` inside a task tells the worker to skip
112+
saving the result to the result backend, so waiting for the result of such a task will time out.
113+
Note that `NoResultError` is still treated as an exception internally, so `on_error` middlewares are called for it as well.
114+
115+
```python
116+
from taskiq import NoResultError
117+
118+
119+
@broker.task
120+
async def my_task() -> None:
121+
raise NoResultError
122+
```

0 commit comments

Comments
 (0)