From 3d514c10649087144278e94c57f5c2fb9c4dca8d Mon Sep 17 00:00:00 2001 From: vvanglro Date: Tue, 4 Aug 2026 18:05:53 +0800 Subject: [PATCH] feat: add skip_result label to skip result backend storage Allow fire-and-forget tasks to opt out of result storage via the skip_result boolean label on @broker.task or kicker.with_labels. --- docs/available-components/result-backends.md | 26 ++++ docs/guide/architecture-overview.md | 9 ++ docs/guide/getting-started.md | 25 ++++ taskiq/kicker.py | 5 +- taskiq/receiver/receiver.py | 43 +++++- tests/receiver/test_receiver.py | 136 +++++++++++++++++++ 6 files changed, 242 insertions(+), 2 deletions(-) diff --git a/docs/available-components/result-backends.md b/docs/available-components/result-backends.md index d5fdcd4a..829fc3e0 100644 --- a/docs/available-components/result-backends.md +++ b/docs/available-components/result-backends.md @@ -10,6 +10,32 @@ This includes: - return value; - Execution time in seconds. +## Skipping result storage + +Sometimes you don't need task results at all (fire-and-forget jobs, notifications, side effects only). +You can skip writing to the result backend for selected tasks with the `skip_result` label: + +```python +@broker.task(skip_result=True) +async def push_notification(user_id: int) -> None: + ... +``` + +Or only for a single call: + +```python +await push_notification.kicker().with_labels(skip_result=True).kiq(user_id=1) +``` + +When `skip_result` is enabled: + +- worker does **not** call `result_backend.set_result`; +- `post_save` middleware hooks are **not** executed; +- callers that use `wait_result()` will time out, because nothing is stored. + +This is complementary to raising `NoResultError` from inside a task. +Use the label when the decision is static; raise `NoResultError` when you decide at runtime. + ## Built-in result backends ### DummyResultBackend diff --git a/docs/guide/architecture-overview.md b/docs/guide/architecture-overview.md index d4b825f1..e1929ded 100644 --- a/docs/guide/architecture-overview.md +++ b/docs/guide/architecture-overview.md @@ -118,6 +118,15 @@ async def main(): ).kiq() ``` +Built-in labels include things like `timeout`, `ack_type`, and `skip_result`. +For example, fire-and-forget tasks can avoid writing to the result backend: + +```python +@broker.task(skip_result=True) +async def send_email(user_id: int) -> None: + ... +``` + Also you can assign custom task names using decorator. This is useful to be sure that task names are unique and resolved correctly. Also it may be useful to balance message routing in some brokers. diff --git a/docs/guide/getting-started.md b/docs/guide/getting-started.md index 2942612d..fcdb3645 100644 --- a/docs/guide/getting-started.md +++ b/docs/guide/getting-started.md @@ -219,6 +219,31 @@ Returned value: 2 Continue reading to get more information about taskiq internals. +## Skipping results + +If a task is fire-and-forget and you do not need `wait_result()`, +set the `skip_result` label so the worker will not store anything in the result backend. + +::: tabs + +@tab decorator + +```python +@broker.task(skip_result=True) +async def push_event(payload: dict) -> None: + ... +``` + +@tab when calling + +```python +await push_event.kicker().with_labels(skip_result=True).kiq(payload={...}) +``` + +::: + +You can also raise `NoResultError` inside a task to skip storage dynamically. + ## Timeouts If you want to restrict amount of time you want to run task, diff --git a/taskiq/kicker.py b/taskiq/kicker.py index b8b9c073..be008893 100644 --- a/taskiq/kicker.py +++ b/taskiq/kicker.py @@ -55,11 +55,14 @@ def __init__( def with_labels( self, - **labels: str | float, + **labels: str | float | bool | bytes, ) -> "AsyncKicker[_FuncParams, _ReturnType]": """ Update function's labels before sending. + Supported value types match label serialization: + ``str``, ``float`` (also accepts ``int``), ``bool`` and ``bytes``. + :param labels: new labels. :return: kicker with new labels. """ diff --git a/taskiq/receiver/receiver.py b/taskiq/receiver/receiver.py index 871d13d6..c73077a0 100644 --- a/taskiq/receiver/receiver.py +++ b/taskiq/receiver/receiver.py @@ -174,8 +174,9 @@ async def callback( # noqa: C901, PLR0912 if middleware.__class__.post_execute != TaskiqMiddleware.post_execute: await maybe_awaitable(middleware.post_execute(taskiq_msg, result)) + should_save = self._should_save_result(taskiq_msg, result) try: - if not isinstance(result.error, NoResultError): + if should_save: await self.broker.result_backend.set_result(taskiq_msg.task_id, result) for middleware in reversed(self.broker.middlewares): @@ -210,6 +211,46 @@ def _get_ack_time(self, message: TaskiqMessage) -> AcknowledgeType: f"Invalid ack_type label {ack_type!r} for task {message.task_name}.", ) from exc + def _should_save_result( + self, + message: TaskiqMessage, + result: TaskiqResult[Any], + ) -> bool: + """ + Decide whether execution result should be stored. + + Results are skipped when: + * task raised ``NoResultError``; + * task has ``skip_result`` label set to ``True``. + """ + if isinstance(result.error, NoResultError): + return False + if self._is_skip_result(message): + logger.debug( + "Task %s with id %s has skip_result label. Skipping result backend.", + message.task_name, + message.task_id, + ) + return False + return True + + @staticmethod + def _is_skip_result(message: TaskiqMessage) -> bool: + """ + Check whether ``skip_result`` label is enabled. + + Only ``True`` and ``False`` are allowed. Missing label means ``False``. + """ + skip_result = message.labels.get("skip_result") + if skip_result is None: + return False + if isinstance(skip_result, bool): + return skip_result + raise ValueError( + f"Invalid skip_result label {skip_result!r} for task " + f"{message.task_name}. Expected True or False.", + ) + async def run_task( # noqa: C901, PLR0912, PLR0915 self, target: Callable[..., Any], diff --git a/tests/receiver/test_receiver.py b/tests/receiver/test_receiver.py index d724b326..6d25eb5b 100644 --- a/tests/receiver/test_receiver.py +++ b/tests/receiver/test_receiver.py @@ -772,6 +772,142 @@ async def task_no_result() -> int: assert not broker._running_tasks +async def test_skip_result_label_on_decorator() -> None: + """Task skip_result label skips result backend and post_save.""" + events: list[str] = [] + broker = ( + InMemoryBroker() + .with_result_backend( + _EventResultBackend(events), + ) + .with_middlewares(_EventMiddleware(events)) + ) + + @broker.task(skip_result=True) + async def my_task() -> int: + events.append("task") + return 1 + + receiver = get_receiver(broker) + broker_message = broker.formatter.dumps(my_task.kicker()._prepare_message()) + + await receiver.callback(broker_message.message) + + assert events == ["task", "post_execute"] + assert not await broker.result_backend.is_result_ready( + broker_message.task_id, + ) + + +async def test_skip_result_label_via_kicker() -> None: + """skip_result can be set per call with kicker labels.""" + events: list[str] = [] + broker = ( + InMemoryBroker() + .with_result_backend( + _EventResultBackend(events), + ) + .with_middlewares(_EventMiddleware(events)) + ) + + @broker.task + async def my_task() -> int: + events.append("task") + return 1 + + receiver = get_receiver(broker) + broker_message = broker.formatter.dumps( + my_task.kicker().with_labels(skip_result=True)._prepare_message(), + ) + + await receiver.callback(broker_message.message) + + assert events == ["task", "post_execute"] + assert not await broker.result_backend.is_result_ready( + broker_message.task_id, + ) + + +async def test_skip_result_false_still_saves_result() -> None: + """Explicit skip_result=False keeps default result storage behavior.""" + events: list[str] = [] + broker = ( + InMemoryBroker() + .with_result_backend( + _EventResultBackend(events), + ) + .with_middlewares(_EventMiddleware(events)) + ) + + @broker.task(skip_result=False) + async def my_task() -> int: + events.append("task") + return 1 + + receiver = get_receiver(broker) + broker_message = broker.formatter.dumps(my_task.kicker()._prepare_message()) + + await receiver.callback(broker_message.message) + + assert events == ["task", "post_execute", "save", "post_save"] + assert await broker.result_backend.is_result_ready(broker_message.task_id) + + +async def test_skip_result_invalid_value_raises() -> None: + """Non-bool skip_result values are rejected.""" + broker = InMemoryBroker() + + @broker.task + async def my_task() -> int: + return 1 + + receiver = get_receiver(broker) + broker_message = broker.formatter.dumps( + TaskiqMessage( + task_id="skip-result-invalid", + task_name=my_task.task_name, + labels={"skip_result": "true"}, + args=[], + kwargs={}, + ), + ) + + with pytest.raises(ValueError, match="Invalid skip_result label"): + await receiver.callback(broker_message.message) + + +async def test_skip_result_still_acks_when_saved() -> None: + """WHEN_SAVED ack still happens if result storage is skipped.""" + events: list[str] = [] + broker = ( + InMemoryBroker() + .with_result_backend( + _EventResultBackend(events), + ) + .with_middlewares(_EventMiddleware(events)) + ) + + @broker.task(skip_result=True) + async def my_task() -> int: + events.append("task") + return 1 + + def ack_callback() -> None: + events.append("ack") + + receiver = get_receiver(broker, ack_type=AcknowledgeType.WHEN_SAVED) + broker_message = broker.formatter.dumps(my_task.kicker()._prepare_message()) + + await receiver.callback( + AckableMessage( + data=broker_message.message, + ack=ack_callback, + ), + ) + + assert events == ["task", "post_execute", "ack"] + + async def test_result() -> None: broker = InMemoryBroker()