Function scheduled_jobs

django_rq/views.py:157–193

(request: HttpRequest, queue_index: int)

@never_cache
@staff_member_required
def scheduled_jobs(request: HttpRequest, queue_index: int) -> HttpResponse:
    queue = get_queue_by_index(queue_index)

    registry = ScheduledJobRegistry(queue.name, queue.connection)

    items_per_page = 100
    num_jobs = len(registry)
    page = int(request.GET.get('page', 1))
    jobs = []

    if request.GET.get('desc', '1') == '1':
        sort_direction = 'descending'
    else:
        sort_direction = 'ascending'

    if num_jobs > 0:
        last_page = int(ceil(num_jobs / items_per_page))
        page_range = list(range(1, last_page + 1))
        offset = items_per_page * (page - 1)
        job_ids = registry.get_job_ids(offset, offset + items_per_page - 1, desc=sort_direction == 'descending')
        jobs = get_jobs(queue, job_ids, registry)
        for job in jobs:
            job.scheduled_at = registry.get_scheduled_time(job)  # type: ignore[attr-defined]
    else:
        page_range = []

    context_data = {
        **each_context(request),
        'queue': queue,
        'queue_index': queue_index,
        'jobs': jobs,
        'num_jobs': num_jobs,
        'page': page,
        'page_range': page_range,
        'sort_direction': sort_direction,
    }
    return render(request, 'django_rq/scheduled_jobs.html', context_data)


@never_cache