Merge branch 'release/0.1.7'

Author: s3rius

docs/extending-taskiq/resutl-backend.md

@@ -14,3 +14,8 @@ Here's a minimal example of a result backend:
 ::: info Cool tip!
 It's a good practice to skip fetching logs from the storage unless `with_logs=True` is explicitly specified.
 :::
+
+
+::: danger Important note!
+`with_logs` param is now deprecated. It will be removed in future releases.
+:::

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "taskiq"
-version = "0.1.6"
+version = "0.1.7"
 description = "Distributed task queue with full async support"
 authors = ["Pavel Kirilin <[email protected]>"]
 maintainers = ["Pavel Kirilin <[email protected]>"]

taskiq/cli/worker/receiver.py

@@ -1,6 +1,5 @@
 import asyncio
 import inspect
-import io
 from concurrent.futures import ThreadPoolExecutor
 from logging import getLogger
 from time import time
@@ -11,7 +10,6 @@
 from taskiq.abc.broker import AsyncBroker
 from taskiq.abc.middleware import TaskiqMiddleware
 from taskiq.cli.worker.args import WorkerArgs
-from taskiq.cli.worker.log_collector import log_collector
 from taskiq.cli.worker.params_parser import parse_params
 from taskiq.context import Context
 from taskiq.message import BrokerMessage, TaskiqMessage
@@ -150,8 +148,6 @@ async def run_task(  # noqa: C901, WPS210
         :return: result of execution.
         """
         loop = asyncio.get_running_loop()
-        # Buffer to capture logs.
-        logs = io.StringIO()
         returned = None
         found_exception = None
         signature = self.task_signatures.get(message.task_name)
@@ -160,55 +156,51 @@ async def run_task(  # noqa: C901, WPS210
             signature = None
         parse_params(signature, self.task_hints.get(message.task_name) or {}, message)
 
-        # Captures function's logs.
-        with log_collector(logs, self.cli_args.log_collector_format):
-            dep_ctx = None
-            if dependency_graph:
-                # Create a context for dependency resolving.
-                dep_ctx = dependency_graph.async_ctx(
-                    {
-                        Context: Context(message, self.broker),
-                        TaskiqState: self.broker.state,
-                    },
-                )
-                # Resolve all function's dependencies.
-                dep_kwargs = await dep_ctx.resolve_kwargs()
-                for key, val in dep_kwargs.items():
-                    if key not in message.kwargs:
-                        message.kwargs[key] = val
-            # Start a timer.
-            start_time = time()
-            try:
-                # If the function is a coroutine we await it.
-                if asyncio.iscoroutinefunction(target):
-                    returned = await target(*message.args, **message.kwargs)
-                else:
-                    # If this is a synchronous function we
-                    # run it in executor.
-                    returned = await loop.run_in_executor(
-                        self.executor,
-                        _run_sync,
-                        target,
-                        message,
-                    )
-            except Exception as exc:
-                found_exception = exc
-                logger.error(
-                    "Exception found while executing function: %s",
-                    exc,
-                    exc_info=True,
+        dep_ctx = None
+        if dependency_graph:
+            # Create a context for dependency resolving.
+            dep_ctx = dependency_graph.async_ctx(
+                {
+                    Context: Context(message, self.broker),
+                    TaskiqState: self.broker.state,
+                },
+            )
+            # Resolve all function's dependencies.
+            dep_kwargs = await dep_ctx.resolve_kwargs()
+            for key, val in dep_kwargs.items():
+                if key not in message.kwargs:
+                    message.kwargs[key] = val
+        # Start a timer.
+        start_time = time()
+        try:
+            # If the function is a coroutine we await it.
+            if asyncio.iscoroutinefunction(target):
+                returned = await target(*message.args, **message.kwargs)
+            else:
+                # If this is a synchronous function we
+                # run it in executor.
+                returned = await loop.run_in_executor(
+                    self.executor,
+                    _run_sync,
+                    target,
+                    message,
                 )
-            # Stop the timer.
-            execution_time = time() - start_time
-            if dep_ctx:
-                await dep_ctx.close()
+        except Exception as exc:
+            found_exception = exc
+            logger.error(
+                "Exception found while executing function: %s",
+                exc,
+                exc_info=True,
+            )
+        # Stop the timer.
+        execution_time = time() - start_time
+        if dep_ctx:
+            await dep_ctx.close()
 
-        raw_logs = logs.getvalue()
-        logs.close()
         # Assemble result.
         result: "TaskiqResult[Any]" = TaskiqResult(
             is_err=found_exception is not None,
-            log=raw_logs,
+            log=None,
             return_value=returned,
             execution_time=execution_time,
         )

taskiq/cli/worker/run.py

@@ -1,9 +1,10 @@
 import asyncio
 import os
 import signal
-from logging import basicConfig, getLevelName, getLogger
-from multiprocessing import Process
-from queue import Queue
+import sys
+from logging import StreamHandler, basicConfig, getLevelName, getLogger
+from logging.handlers import QueueHandler, QueueListener
+from multiprocessing import Process, Queue
 from time import sleep
 from typing import Any, List
 
@@ -223,10 +224,14 @@ def run_worker(args: WorkerArgs) -> None:  # noqa: WPS213
 
     :param args: CLI arguments.
     """
+    logging_queue = Queue(-1)  # type: ignore
+    listener = QueueListener(logging_queue, StreamHandler(sys.stdout))
     basicConfig(
         level=getLevelName(args.log_level),
-        format=("[%(asctime)s][%(levelname)-7s][%(processName)s] %(message)s"),
+        format="[%(asctime)s][%(levelname)-7s][%(processName)s] %(message)s",
+        handlers=[QueueHandler(logging_queue)],
     )
+    listener.start()
     logger.info("Starting %s worker processes.", args.workers)
 
     global worker_processes  # noqa: WPS420
@@ -259,3 +264,4 @@ def run_worker(args: WorkerArgs) -> None:  # noqa: WPS213
     signal.signal(signal.SIGTERM, signal_handler)
 
     watcher_loop(args=args)
+    listener.stop()

taskiq/result.py

@@ -9,6 +9,9 @@ class TaskiqResult(GenericModel, Generic[_ReturnType]):
     """Result of a remote task invocation."""
 
     is_err: bool
+    # Log is a deprecated field. It would be removed in future
+    # releases of not, if we find a way to capture logs in async
+    # environment.
     log: Optional[str]
     return_value: _ReturnType
     execution_time: float