Function get_worker

django_rq/workers.py:42–64

Returns a RQ worker for all queues or specified ones.

(
    *queue_names: str,
    job_class: Optional[Union[str, type[Job]]] = None,
    queue_class: Optional[Union[str, type[DjangoRQ]]] = None,
    worker_class: Optional[Union[str, type[Worker]]] = None,
    **kwargs,
)

def get_worker(
    *queue_names: str,
    job_class: Optional[Union[str, type[Job]]] = None,
    queue_class: Optional[Union[str, type[DjangoRQ]]] = None,
    worker_class: Optional[Union[str, type[Worker]]] = None,
    **kwargs,
) -> Worker:
    """
    Returns a RQ worker for all queues or specified ones.
    """
    job_class = get_job_class(job_class)
    queues = get_queues(*queue_names, job_class=job_class, queue_class=queue_class)
    # normalize queue_class to what get_queues returns
    queue_class = queues[0].__class__
    worker_class = get_worker_class(worker_class)
    return worker_class(
        queues,
        connection=queues[0].connection,
        exception_handlers=get_exception_handlers() or None,
        job_class=job_class,
        queue_class=queue_class,
        **kwargs,
    )