From b31f429f907d74c72e8807661a97e71f5d910934 Mon Sep 17 00:00:00 2001 From: Dmitry Tatarkin Date: Tue, 4 Nov 2025 20:20:02 +0300 Subject: [PATCH] feat: add ability to specify the default WorkerArgs when parsing from the command line using the WorkerArgs.from_cli() method. --- taskiq/cli/worker/args.py | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/taskiq/cli/worker/args.py b/taskiq/cli/worker/args.py index ca44e9d..9a222e0 100644 --- a/taskiq/cli/worker/args.py +++ b/taskiq/cli/worker/args.py @@ -1,6 +1,6 @@ -from argparse import ArgumentDefaultsHelpFormatter, ArgumentParser +from argparse import ArgumentDefaultsHelpFormatter, ArgumentParser, Namespace from dataclasses import dataclass, field -from typing import List, Optional, Sequence, Tuple +from typing import Any, Dict, List, Optional, Sequence, Tuple from taskiq.acks import AcknowledgeType from taskiq.cli.common_args import LogLevel @@ -58,11 +58,13 @@ class WorkerArgs: def from_cli( cls, args: Optional[Sequence[str]] = None, + defaults: Optional[Dict[str, Any]] = None, ) -> "WorkerArgs": """ Construct TaskiqArgs instanc from CLI arguments. :param args: list of args as for cli. + :param defaults: default worker arguments. :return: TaskiqArgs instance. """ parser = ArgumentParser(formatter_class=ArgumentDefaultsHelpFormatter) @@ -267,7 +269,10 @@ def from_cli( help="Maximum number of processes in process pool.", ) - namespace = parser.parse_args(args) + namespace = parser.parse_args( + args, + namespace=None if defaults is None else Namespace(**defaults), + ) # If there are any patterns specified, remove default. # This is an argparse limitation. if len(namespace.tasks_pattern) > 1: