feat: receiver cli args (#124)

Co-authored-by: Anton <[email protected]>

Author: Sobes76rus

taskiq/cli/worker/args.py

@@ -1,10 +1,24 @@
 from argparse import ArgumentDefaultsHelpFormatter, ArgumentParser
-from dataclasses import dataclass
-from typing import List, Optional, Sequence
+from dataclasses import dataclass, field
+from typing import List, Optional, Sequence, Tuple
 
 from taskiq.cli.common_args import LogLevel
 
 
+def receiver_arg_type(string: str) -> Tuple[str, str]:
+    """
+    Parse cli --receiver_arg argument value.
+
+    :param string: cli argument value in format key=value.
+    :raises ValueError: if value not in format.
+    :return: (key, value) pair.
+    """
+    args = string.split("=", 1)
+    if len(args) != 2:
+        raise ValueError(f"Invalid value: {string}")
+    return args[0], args[1]
+
+
 @dataclass
 class WorkerArgs:
     """Taskiq worker CLI arguments."""
@@ -25,6 +39,7 @@ class WorkerArgs:
     no_gitignore: bool = False
     max_async_tasks: int = 100
     receiver: str = "taskiq.receiver:Receiver"
+    receiver_arg: List[Tuple[str, str]] = field(default_factory=list)
 
     @classmethod
     def from_cli(  # noqa: WPS213
@@ -55,6 +70,17 @@ def from_cli(  # noqa: WPS213
                 "'module.module:variable' format."
             ),
         )
+        parser.add_argument(
+            "--receiver_arg",
+            action="append",
+            type=receiver_arg_type,
+            default=[],
+            help=(
+                "List of args fot receiver. "
+                "This string must be specified in "
+                "`key=value` format."
+            ),
+        )
         parser.add_argument(
             "--tasks-pattern",
             "-tp",

taskiq/cli/worker/run.py

@@ -93,6 +93,7 @@ def start_listen(args: WorkerArgs) -> None:  # noqa: WPS210, WPS213
         raise ValueError("Unknown broker type. Please use AsyncBroker instance.")
 
     receiver_type = get_receiver_type(args)
+    receiver_args = dict(args.receiver_arg)
 
     # Here how we manage interruptions.
     # We have to remember shutting_down state,
@@ -131,6 +132,7 @@ def interrupt_handler(signum: int, _frame: Any) -> None:
                 executor=pool,
                 validate_params=not args.no_parse,
                 max_async_tasks=args.max_async_tasks,
+                **receiver_args,
             )
             loop.run_until_complete(receiver.listen())
     except KeyboardInterrupt: