Function failed_jobs

django_rq/views.py:116–152

(request: HttpRequest, queue_index: int)

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

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

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

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

    jobs = []

    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)

    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/failed_jobs.html', context_data)


@never_cache