CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-django-celery-results

Celery Result Backends using the Django ORM/Cache framework

—

Pending
Overview
Eval results
Files

views.mddocs/

REST API Views

HTTP endpoints for checking task and group execution status, returning JSON responses suitable for web applications and AJAX requests. These views provide programmatic access to task status without requiring direct database queries.

Core Imports

from django_celery_results.views import (
    task_status,
    is_task_successful, 
    group_status,
    is_group_successful
)

Capabilities

Task Status Views

Views for checking individual task execution status and retrieving detailed results.

def task_status(request, task_id):
    """
    Get detailed task status and result in JSON format.
    
    Args:
        request: Django HttpRequest object
        task_id (str): Celery task ID to check
    
    Returns:
        JsonResponse: JSON with task status, result, and error information
            {
                "task": {
                    "id": "task-id",
                    "status": "SUCCESS|PENDING|FAILURE|...",
                    "result": "task result data",
                    # For failed tasks:
                    "exc": "exception class name",
                    "traceback": "full traceback string"
                }
            }
    """

def is_task_successful(request, task_id):
    """
    Check if a task completed successfully.
    
    Args:
        request: Django HttpRequest object
        task_id (str): Celery task ID to check
    
    Returns:
        JsonResponse: JSON with boolean success status
            {
                "task": {
                    "id": "task-id",
                    "executed": true|false
                }
            }
    """

Usage Examples

# URL configuration
from django.urls import path, include

urlpatterns = [
    path('celery/', include('django_celery_results.urls')),
]

# JavaScript/AJAX usage
fetch('/celery/task/status/abc123/')
    .then(response => response.json())
    .then(data => {
        console.log('Task status:', data.task.status);
        console.log('Task result:', data.task.result);
        if (data.task.status === 'FAILURE') {
            console.log('Error:', data.task.exc);
            console.log('Traceback:', data.task.traceback);
        }
    });

// Check if task succeeded
fetch('/celery/task/done/abc123/')
    .then(response => response.json())
    .then(data => {
        if (data.task.executed) {
            console.log('Task completed successfully');
        } else {
            console.log('Task not yet completed or failed');
        }
    });

Group Status Views

Views for checking group task execution status and retrieving results for all tasks in a group.

def group_status(request, group_id):
    """
    Get detailed status for all tasks in a group.
    
    Args:
        request: Django HttpRequest object
        group_id (str): Celery group ID to check
    
    Returns:
        JsonResponse: JSON with group ID and all task results
            {
                "group": {
                    "id": "group-id",
                    "results": [
                        {
                            "result": "task 1 result",
                            "status": "SUCCESS"
                        },
                        {
                            "result": "task 2 result", 
                            "status": "FAILURE"
                        }
                    ]
                }
            }
    """

def is_group_successful(request, group_id):
    """
    Check if all tasks in a group completed successfully.
    
    Args:
        request: Django HttpRequest object
        group_id (str): Celery group ID to check
    
    Returns:
        JsonResponse: JSON with group ID and per-task execution status
            {
                "group": {
                    "id": "group-id",
                    "results": [
                        {
                            "id": "task-1-id",
                            "executed": true
                        },
                        {
                            "id": "task-2-id", 
                            "executed": false
                        }
                    ]
                }
            }
    """

Usage Examples

# Check detailed group status
fetch('/celery/group/status/group123/')
    .then(response => response.json())
    .then(data => {
        console.log('Group ID:', data.group.id);
        data.group.results.forEach((task, index) => {
            console.log(`Task ${index}: ${task.status} - ${task.result}`);
        });
    });

// Check if all group tasks succeeded
fetch('/celery/group/done/group123/')
    .then(response => response.json())
    .then(data => {
        const allSuccessful = data.group.results.every(task => task.executed);
        if (allSuccessful) {
            console.log('All group tasks completed successfully');
        } else {
            console.log('Some group tasks are still pending or failed');
            data.group.results.forEach(task => {
                if (!task.executed) {
                    console.log(`Task ${task.id} not completed`);
                }
            });
        }
    });

URL Configuration

URL Patterns

The package provides predefined URL patterns for all views:

# django_celery_results.urls
urlpatterns = [
    # Task status endpoints
    path('task/status/<task_pattern:task_id>/', views.task_status, 
         name='celery-task_status'),
    path('task/done/<task_pattern:task_id>/', views.is_task_successful, 
         name='celery-is_task_successful'),
    
    # Group status endpoints  
    path('group/status/<task_pattern:group_id>/', views.group_status, 
         name='celery-group_status'),
    path('group/done/<task_pattern:group_id>/', views.is_group_successful, 
         name='celery-is_group_successful'),
]

Custom Path Converter

A custom path converter handles task and group ID patterns:

class TaskPatternConverter:
    """Custom path converter for task and group IDs."""
    
    regex: str  # Regular expression pattern: r'[\w\d\-\.]+'
    
    def to_python(self, value):
        """
        Convert URL value to Python string.
        
        Args:
            value: URL path component
            
        Returns:
            str: Task or group ID as string
        """
    
    def to_url(self, value):
        """
        Convert Python value to URL component.
        
        Args:
            value: Task or group ID
            
        Returns:
            str: URL-safe string
        """

Legacy URL Support

The package includes deprecated ID-first URL patterns for backward compatibility:

# Legacy URLs (deprecated, controlled by setting)
DJANGO_CELERY_RESULTS_ID_FIRST_URLS = True  # Default

# Legacy patterns (will be removed in future versions):
# /<task_id>/done/
# /<task_id>/status/
# /<group_id>/group/done/
# /<group_id>/group/status/

Integration Examples

Django Template Usage

<!-- task_status.html -->
<div id="task-status" data-task-id="{{ task_id }}">
    <p>Status: <span id="status">Loading...</span></p>
    <p>Result: <span id="result">-</span></p>
    <div id="error-info" style="display: none;">
        <p>Error: <span id="error-type"></span></p>
        <pre id="traceback"></pre>
    </div>
</div>

<script>
function checkTaskStatus(taskId) {
    fetch(`/celery/task/status/${taskId}/`)
        .then(response => response.json())
        .then(data => {
            document.getElementById('status').textContent = data.task.status;
            
            if (data.task.status === 'SUCCESS') {
                document.getElementById('result').textContent = data.task.result;
            } else if (data.task.status === 'FAILURE') {
                document.getElementById('error-type').textContent = data.task.exc;
                document.getElementById('traceback').textContent = data.task.traceback;
                document.getElementById('error-info').style.display = 'block';
            }
        });
}

// Poll for status updates
const taskId = document.getElementById('task-status').dataset.taskId;
const interval = setInterval(() => {
    checkTaskStatus(taskId);
}, 2000);
</script>

React Component Usage

// TaskStatus.jsx
import React, { useState, useEffect } from 'react';

function TaskStatus({ taskId }) {
    const [status, setStatus] = useState(null);
    const [loading, setLoading] = useState(true);

    useEffect(() => {
        const checkStatus = async () => {
            try {
                const response = await fetch(`/celery/task/status/${taskId}/`);
                const data = await response.json();
                setStatus(data.task);
                setLoading(false);
                
                // Stop polling if task is complete
                if (['SUCCESS', 'FAILURE', 'REVOKED'].includes(data.task.status)) {
                    return;
                }
            } catch (error) {
                console.error('Error checking task status:', error);
                setLoading(false);
            }
        };

        checkStatus();
        const interval = setInterval(checkStatus, 2000);
        
        return () => clearInterval(interval);
    }, [taskId]);

    if (loading) return <div>Loading...</div>;

    return (
        <div>
            <h3>Task Status: {status.status}</h3>
            {status.status === 'SUCCESS' && (
                <div>Result: {JSON.stringify(status.result)}</div>
            )}
            {status.status === 'FAILURE' && (
                <div>
                    <div>Error: {status.exc}</div>
                    <pre>{status.traceback}</pre>
                </div>
            )}
        </div>
    );
}

Response Formats

Task Status Response

{
    "task": {
        "id": "550e8400-e29b-41d4-a716-446655440000",
        "status": "SUCCESS",
        "result": "Task completed successfully"
    }
}

Failed Task Response

{
    "task": {
        "id": "550e8400-e29b-41d4-a716-446655440000",
        "status": "FAILURE",
        "result": "ValueError: Invalid input",
        "exc": "ValueError",
        "traceback": "Traceback (most recent call last):\n  File ...\nValueError: Invalid input"
    }
}

Group Status Response

{
    "group": {
        "id": "group-550e8400-e29b-41d4-a716-446655440000",
        "results": [
            {
                "result": "First task result",
                "status": "SUCCESS"
            },
            {
                "result": "Second task result",
                "status": "SUCCESS"
            }
        ]
    }
}

Install with Tessl CLI

npx tessl i tessl/pypi-django-celery-results

docs

admin.md

backends.md

index.md

models.md

views.md

tile.json