taskiq
taskiq copied to clipboard
feat: add ability to specify the default WorkerArgs …
… when parsing from the command line using the WorkerArgs.from_cli() method.
This may be useful if you want to implement a custom command-line interface (CLI) command to run project-specific workers with custom default arguments.
Here is how I use Taskiq with Django:
import logging
import os
from collections.abc import Sequence
import configurations
from taskiq.abc.cmd import TaskiqCMD
from taskiq.cli.worker.args import WorkerArgs
from taskiq.cli.worker.run import run_worker
logger = logging.getLogger(__name__)
class ProjectWorkerCMD(TaskiqCMD):
short_help = "Run Taskiq worker"
def exec(self, args: Sequence[str]) -> None:
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "project.settings.web")
os.environ.setdefault("DJANGO_CONFIGURATION", "Web")
configurations.setup()
worker_args = WorkerArgs.from_cli(args, defaults=dict(
fs_discover=True,
app_dir="src/project/root/",
tasks_pattern=['**/taskiq/tasks.py'],
))
run_worker(args=worker_args)