or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

admin.mdbackends.mdindex.mdmodels.mdviews.md
tile.json

tessl/pypi-django-celery-results

Celery Result Backends using the Django ORM/Cache framework

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/django-celery-results@2.6.x

To install, run

npx @tessl/cli install tessl/pypi-django-celery-results@2.6.0

index.mddocs/

Django Celery Results

A Django extension providing database and cache backends for storing Celery task results. It enables developers to store and retrieve Celery task results using Django's ORM models and caching framework, allowing task results to be queried like any other Django model.

Package Information

  • Package Name: django-celery-results
  • Language: Python
  • Installation: pip install django-celery-results

Core Imports

import django_celery_results

Backend imports for configuration:

from django_celery_results.backends import DatabaseBackend, CacheBackend

Model imports for querying results:

from django_celery_results.models import TaskResult, GroupResult, ChordCounter

App configuration import:

from django_celery_results.apps import CeleryResultConfig

Utility function imports:

from django_celery_results.utils import now

Basic Usage

Installation and Configuration

# settings.py
INSTALLED_APPS = [
    # ... other apps
    'django_celery_results',
]

# Configure Celery to use Django database backend
CELERY_RESULT_BACKEND = 'django-db'

# Or use Django cache backend
# CELERY_RESULT_BACKEND = 'django-cache'

# Optional: Configure task ID max length (useful for MySQL)
DJANGO_CELERY_RESULTS_TASK_ID_MAX_LENGTH = 191

Querying Task Results

from django_celery_results.models import TaskResult
from celery.result import AsyncResult

# Query task results using Django ORM
recent_tasks = TaskResult.objects.filter(
    status='SUCCESS',
    date_done__gte=timezone.now() - timedelta(hours=1)
)

for task in recent_tasks:
    print(f"Task {task.task_id}: {task.status}")
    print(f"Result: {task.result}")

# Use Celery's AsyncResult with the backend
result = AsyncResult('task-id-here')
print(f"Task state: {result.state}")
print(f"Task result: {result.result}")

Architecture

Django Celery Results provides two main backend implementations:

  • DatabaseBackend: Stores task results in Django database tables using the ORM
  • CacheBackend: Stores task results in Django's cache framework
  • Models: Django models (TaskResult, GroupResult, ChordCounter) for database storage
  • Managers: Custom managers providing database operations with transaction retry logic
  • Admin Integration: Django admin interfaces for browsing and managing task results
  • REST Views: HTTP endpoints for checking task and group status via JSON APIs

The package integrates seamlessly with Celery's result backend system while providing the full power of Django's ORM for querying and managing task results.

Capabilities

Result Backend Configuration

Configuration of Celery result backends using Django's database or cache frameworks. Provides both database-based and cache-based storage with automatic encoding/decoding of task results.

class DatabaseBackend(BaseDictBackend):
    def __init__(self, *args, **kwargs): ...
    def exception_safe_to_retry(self, exc): ...
    def encode_content(self, data): ...
    def decode_content(self, obj, content): ...
    def cleanup(self): ...

class CacheBackend(KeyValueStoreBackend):
    def __init__(self, *args, **kwargs): ...
    def get(self, key): ...
    def set(self, key, value): ...
    def delete(self, key): ...

Backend Configuration

Task Result Models

Django models for storing and querying Celery task results, group results, and chord coordination data. Provides full ORM capabilities for result management and includes custom managers with retry logic.

class TaskResult(models.Model):
    task_id: models.CharField
    task_name: models.CharField
    status: models.CharField
    result: models.TextField
    date_created: models.DateTimeField
    date_done: models.DateTimeField
    
    def as_dict(self): ...

class GroupResult(models.Model):
    group_id: models.CharField
    result: models.TextField
    date_created: models.DateTimeField
    date_done: models.DateTimeField
    
    def as_dict(self): ...

class ChordCounter(models.Model):
    group_id: models.CharField
    sub_tasks: models.TextField
    count: models.PositiveIntegerField
    
    def group_result(self, app=None): ...

Task Result Models

Admin Interface

Django admin integration for browsing, searching, and managing task results through the web interface. Provides read-only or editable views based on configuration.

class TaskResultAdmin(admin.ModelAdmin):
    model: TaskResult
    list_display: tuple
    list_filter: tuple
    search_fields: tuple
    
    def get_readonly_fields(self, request, obj=None): ...

class GroupResultAdmin(admin.ModelAdmin):
    model: GroupResult
    list_display: tuple
    search_fields: tuple

Admin Interface

REST API Views

HTTP endpoints for checking task and group execution status, returning JSON responses suitable for web applications and AJAX requests.

def task_status(request, task_id):
    """Get detailed task status and result in JSON format."""
    
def is_task_successful(request, task_id):
    """Check if a task completed successfully."""
    
def group_status(request, group_id):
    """Get detailed status for all tasks in a group."""
    
def is_group_successful(request, group_id):
    """Check if all tasks in a group completed successfully."""

REST API Views

Django App Configuration

Django application configuration class providing package metadata and settings for the django-celery-results app.

class CeleryResultConfig(AppConfig):
    """Default configuration for the django_celery_results app."""

    name: str  # 'django_celery_results'
    label: str  # 'django_celery_results'  
    verbose_name: str  # 'Celery Results'
    default_auto_field: str  # 'django.db.models.AutoField'

Utility Functions

Helper functions for time handling that work correctly with Django timezone settings.

def now():
    """
    Return the current date and time with proper timezone handling.
    
    Returns:
        datetime: Current timestamp respecting Django USE_TZ setting
    """