Merge branch 'release/0.3.0'

Author: s3rius

docs/available-components/brokers.md

@@ -79,22 +79,34 @@ await my_task.kicker().with_broker(broker).kiq()
 
 :::
 
-## AioPikaBroker (for RabbitMQ)
 
-This broker is not part of the core taskiq lib. You can install it as a separate package [taskiq-aio-pika](https://pypi.org/project/taskiq-aio-pika/).
+## Custom brokers
+
+These brokers are not parts of the core taskiq lib. You can install them as a separate packages.
+
+You can read more about parameters and abilities of these brokers in README.md of each repo.
+
+
+###  AioPikaBroker (for RabbitMQ)
+
+Project link: [taskiq-aio-pika](https://github.com/taskiq-python/taskiq-aio-pika).
 
 ```bash
 pip install taskiq-aio-pika
 ```
 
-You can read more about parameters and abilities of this broker in [README.md](https://github.com/taskiq-python/taskiq-aio-pika).
+### Redis broker
 
-## Redis broker
-
-This broker is not part of the core taskiq lib. You can install it as a separate package [taskiq-redis](https://pypi.org/project/taskiq-redis/).
+Project link: [taskiq-redis](https://github.com/taskiq-python/taskiq-redis).
 
 ```bash
 pip install taskiq-redis
 ```
 
-You can read more about parameters and abilities of this broker in [README.md](https://github.com/taskiq-python/taskiq-redis).
+### NATS broker
+
+Project link: [taskiq-nats](https://github.com/taskiq-python/taskiq-nats).
+
+```bash
+pip install taskiq-nats
+```

docs/guide/state-and-deps.md

@@ -171,3 +171,13 @@ By default taskiq has only two deendencies:
 
 - Context from `taskiq.context.Context`
 - TaskiqState from `taskiq.state.TaskiqState`
+
+
+### Adding first-level dependencies
+
+You can expand default list of available dependencies for you application.
+Taskiq have an ability to add new first-level dependencies using brokers.
+
+The AsyncBroker interface has a function called `add_dependency_context` and you can add
+more default dependencies to the taskiq. This may be useful for libraries if you want to
+add new dependencies to users.

poetry.lock

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "taskiq"
-version = "0.2.2"
+version = "0.3.0"
 description = "Distributed task queue with full async support"
 authors = ["Pavel Kirilin <win10@list.ru>"]
 maintainers = ["Pavel Kirilin <win10@list.ru>"]
@@ -35,7 +35,7 @@ watchdog = "^2.1.9"
 gitignore-parser = "^0.1.0"
 importlib-metadata = "<4.3"
 pycron = "^3.0.0"
-taskiq_dependencies =  "~1.0.0"
+taskiq_dependencies =  "^1.0.0"
 
 [tool.poetry.dev-dependencies]
 pytest = "^7.1.2"

pyproject.toml

@@ -86,11 +86,23 @@ def __init__(
         # Every event has a list of handlers.
         # Every handler is a function which takes state as a first argument.
         # And handler can be either sync or async.
-        self.event_handlers: DefaultDict[  # noqa: WPS234
+        self.event_handlers: DefaultDict[
             TaskiqEvents,
             List[Callable[[TaskiqState], Optional[Awaitable[None]]]],
         ] = defaultdict(list)
         self.state = TaskiqState()
+        self.custom_dependency_context: Dict[Any, Any] = {}
+
+    def add_dependency_context(self, new_ctx: Dict[Any, Any]) -> None:
+        """
+        Add first-level dependencies.
+
+        Provided dict will be used to inject new dependencies
+        in all dependency graph contexts.
+
+        :param new_ctx: Additional context values for dependnecy injection.
+        """
+        self.custom_dependency_context.update(new_ctx)
 
     def add_middlewares(self, *middlewares: "TaskiqMiddleware") -> None:
         """

taskiq/abc/broker.py

@@ -159,12 +159,14 @@ async def run_task(  # noqa: C901, WPS210
         dep_ctx = None
         if dependency_graph:
             # Create a context for dependency resolving.
-            dep_ctx = dependency_graph.async_ctx(
+            broker_ctx = self.broker.custom_dependency_context
+            broker_ctx.update(
                 {
                     Context: Context(message, self.broker),
                     TaskiqState: self.broker.state,
                 },
             )
+            dep_ctx = dependency_graph.async_ctx(broker_ctx)
             # Resolve all function's dependencies.
             dep_kwargs = await dep_ctx.resolve_kwargs()
             for key, val in dep_kwargs.items():

taskiq/cli/worker/receiver.py

@@ -1,6 +1,7 @@
 from typing import Any, Optional
 
 import pytest
+from taskiq_dependencies import Depends
 
 from taskiq.abc.broker import AsyncBroker
 from taskiq.abc.middleware import TaskiqMiddleware
@@ -200,3 +201,43 @@ async def test_callback_unknown_task() -> None:
     )
 
     await receiver.callback(broker_message)
+
+
+@pytest.mark.anyio
+async def test_custom_ctx() -> None:
+    """Tests that run_task can run sync tasks."""
+
+    class MyTestClass:
+        """Class to test injection."""
+
+        def __init__(self, val: int) -> None:
+            self.val = val
+
+    broker = InMemoryBroker()
+
+    # We register a task into broker,
+    # to build dependency graph on startup.
+    @broker.task
+    def test_func(tes_val: MyTestClass = Depends()) -> int:
+        return tes_val.val
+
+    # We add custom first-level dependency.
+    broker.add_dependency_context({MyTestClass: MyTestClass(11)})
+    # Create a receiver.
+    receiver = get_receiver(broker)
+
+    result = await receiver.run_task(
+        test_func,
+        TaskiqMessage(
+            task_id="",
+            task_name=test_func.task_name,
+            labels={},
+            args=[],
+            kwargs={},
+        ),
+    )
+
+    # Check that the value is equal
+    # to the one we supplied.
+    assert result.return_value == 11
+    assert not result.is_err