blog about dark mmfilesi
python · ·12 min ·Intermediate

Introducción a Celery II

Segunda parte de la introducción a Celery, en la que veremos cómo se integra con Django

#celery#python#djanjo#broker

3. Integración con Django

3.1. Preparación

En esta segunda entrada dedicada a Celery vamos a ver cómo funciona el backend de resultados, los estados de una tarea, qué sucede cuando falla una y qué son los reintentos automáticos, para lo cual vamos a apoyarnos en un servicio en Django que nos sirva de productor, de lanzador de tareas. Esta pieza es la de menos en este tutorial, puedes utilizar cualquier otro framework que sirva para preparar un endpoint o, incluso, directamente en vanilla.

Lo primero es preparar un endpoint en Django o whatever que de momento no tenga nada.

from django.http import JsonResponse
from django.views.decorators.http import require_GET


@require_GET
def foo(request):
    return JsonResponse({'foo': 'bar'})

Instalamos Celery con las dependencias necesarias para utilizar Redis:

uv add "celery[redis]"

Y levantamos un Redis como broker con Docker en el caso de que aún no lo tengamos de la entrada anterior.

docker run \
  --name celery-redis \
  -p 127.0.0.1:6379:6379 \
  redis:7-alpine

Si ya lo tenemos, lo arrancamos sin más

docker start celery-redis

3.2. Configuración de Celery

Una vez tenemos Django y Redis funcionando, debemos crear y configurar la aplicación de Celery que utilizará nuestro proyecto.

Supongamos que la estructura inicial es la siguiente:


config/
  - __init__.py
  - settings.py
  - urls.py
  - wsgi.py

api/ <--- o el nombre que quieras poner a una app del proyecto
  - urls.py
  - views.py
  - archivillos varios

manage.py

config es el paquete principal del proyecto Django y api es la aplicación en la que tenemos nuestro endpoint, que puedes llamar como quieras.

3.2.1. celery.py

Dentro de config creamos un archivo llamado celery.py:

config/
  - __init__.py
  - settings.py
  - urls.py
  - wsgi.py
  - celery.py

Y añadimos la configuración de Celery:

import os

from celery import Celery


os.environ.setdefault(
    "DJANGO_SETTINGS_MODULE",
    "config.settings",
)

app = Celery("config")

app.config_from_object(
    "django.conf:settings",
    namespace="CELERY",
)

app.autodiscover_tasks()

Vamos por partes.

Primero indicamos a Celery dónde se encuentra la configuración de Django:

os.environ.setdefault(
    "DJANGO_SETTINGS_MODULE",
    "config.settings",
)

Es la misma variable que utilizan manage.py, wsgi.py y asgi.py para localizar el archivo settings.py.

Después creamos la aplicación de Celery:

app = Celery("config")

El nombre config identifica nuestra aplicación Celery. Podríamos utilizar otro, pero normalmente se usa el mismo nombre que tiene el proyecto Django.

A continuación, le pedimos que cargue su configuración desde los ajustes de Django:

app.config_from_object(
    "django.conf:settings",
    namespace="CELERY",
)

El parámetro namespace="CELERY" significa que Celery solamente tendrá en cuenta las variables de settings.py cuyo nombre comience por CELERY_.

Por ejemplo:

CELERY_BROKER_URL = "redis://localhost:6379/0"

Celery interpretará esta variable como su opción broker_url.

Finalmente, utilizamos:

app.autodiscover_tasks()

Con esta instrucción Celery recorrerá las aplicaciones instaladas en Django y buscará módulos llamados tasks.py. Las tareas que definamos en esos archivos serán registradas automáticamente cuando arranque el worker. Es decir, cuando arranque la aplicación buscará en todos los directorios de las apps el archivo tasks.py para auto-importar las tareas celery. Si no lo hiciéramos así, tendríamos que importarlas una a una.

from app1.tasks import mi_tarea1
from app2.tasks import mi_tarea2
# ... importamos cada tarea de cada app manualmente

Y si quisiéramos que se llamaran de otra forma en vez de tasks podríamos definírselo en parámetro related_name.

app.autodiscover_tasks(related_name="otro_nombre_fantastico")

3.2.2. settings e init

Luego añadimos la dirección del broker al archivo config/settings.py:

CELERY_BROKER_URL = "redis://localhost:6379/0"

Por el momento solo estamos configurando el broker. El backend de resultados lo veremos en la sección 5.

Por último, abrimos config/__init__.py y añadimos:

from .celery import app as celery_app


__all__ = ("celery_app",)

A primera vista, este código puede resultar extraño. Ya hemos creado la aplicación de Celery en config/celery.py, así que ¿por qué necesitamos importarla otra vez? Para entenderlo debemos detenernos un momento en la función de __init__.py.

__init__.py es un archivo especial de Python asociado a los paquetes. Las versiones modernas de Python también permiten ciertos paquetes sin __init__.py, llamados namespace packages, pero los proyectos Django continúan creándolo de manera predeterminada.

Lo importante para nosotros es que el contenido de __init__.py se ejecuta cuando Python importa el paquete. Si el archivo está vacío, no sucede nada más. Por eso muchas veces parece que __init__.py no sirve para nada. Sin embargo, podemos utilizarlo para ejecutar código de inicialización o exponer determinados objetos del paquete.

Normalmente, su contenido se ejecuta una sola vez por proceso. Después de la primera importación, Python conserva el módulo cargado y las siguientes importaciones reutilizan esa misma instancia.

Entonces, en config/celery.py hemos creado la aplicación de Celery:

app = Celery("config")

Pero crear un archivo no significa que Python vaya a ejecutarlo automáticamente. Para que se ejecute el contenido de un módulo, alguien debe importarlo.

Django sabe localizar determinados componentes de su propia arquitectura, como:

  • settings.py;
  • urls.py;
  • wsgi.py;
  • asgi.py;
  • las aplicaciones incluidas en INSTALLED_APPS.

Sin embargo, celery.py no es un archivo especial para Django. Si nadie lo importa, Django no tiene por qué abrirlo y este código podría no llegar a ejecutarse:

app = Celery("config")

Por eso lo importamos desde config/__init__.py:

from .celery import app as celery_app

El punto inicial de .celery es una importación relativa y significa: "importa el módulo celery.py que se encuentra dentro de este mismo paquete".

Cuando Django carga su configuración, Python importa el paquete config para poder acceder a config.settings. Al importar config, se ejecuta su __init__.py; este importa config/celery.py y, como consecuencia, se crea y configura la aplicación de Celery.

El recorrido completo es el siguiente:

Django necesita config.settings >
Python importa el paquete config >
Ejecuta config/__init__.py >
config/__init__.py importa config/celery.py >
Se ejecuta app = Celery("config") >
Celery carga la configuración de Django >
Celery busca las tareas registradas

Por tanto, no estamos creando una segunda aplicación de Celery. Estamos forzando la carga de la única instancia que ya hemos definido en config/celery.py.

3.2.3. La relación con @shared_task

Como veremos, esta importación también es importante cuando declaramos tareas mediante @shared_task.

Dentro de una aplicación Django crearemos tareas como esta:

from celery import shared_task


@shared_task
def saludar(nombre):
    return f"Hola, {nombre}"

Observa que este archivo no importa directamente nuestra instancia:

from config.celery import app

En su lugar, utiliza el decorador genérico @shared_task. Este decorador asocia la tarea con la aplicación de Celery que esté activa cuando se carguen las tareas.

Por eso necesitamos que nuestra instancia se haya cargado y configurado correctamente antes de que Django importe los módulos tasks.py.

La importación desde __init__.py garantiza este orden:

  1. Se importa el proyecto Django
  2. Se crea y configura la aplicación Celery
  3. Celery descubre los módulos tasks.py
  4. @shared_task utiliza la aplicación configurada

Sin esta importación, @shared_task podría no utilizar la aplicación de Celery que hemos configurado para nuestro proyecto.

La segunda línea de config/__init__.py es:

__all__ = ("celery_app",)

__all__ define qué nombres considera públicos el paquete cuando se realiza una importación con asterisco:

from config import *

En este caso, indica que el paquete expone públicamente celery_app.

No obstante, __all__ no carga Celery y no es imprescindible para que la integración funcione. La línea que realmente provoca la carga de la aplicación es:

from .celery import app as celery_app

Podríamos dejar el archivo así:

from .celery import app as celery_app

y Celery seguiría funcionando. La documentación añade __all__ para declarar explícitamente que celery_app forma parte de la interfaz pública del paquete.

En definitiva, el contenido de config/__init__.py cumple dos funciones:

from .celery import app as celery_app

Importa y carga la aplicación de Celery cuando Python importa el paquete principal del proyecto Django.

__all__ = ("celery_app",)

Declara que esa aplicación es uno de los objetos públicos que expone el paquete.

Con esto nos aseguramos de que la aplicación de Celery está configurada antes de que se registren las tareas y de que los decoradores @shared_task utilizan la instancia correcta.

4. Crear una tarea en Django

4.1. La tarea de Celery

Como acabamos de ver, Celery va a buscar automáticamente módulos llamados tasks.py dentro de las aplicaciones declaradas en INSTALLED_APPS, así que en nuestro caso vamos a crear este archivo en el directorio donde tengamos la app de Django, el que he llamado api en este tutorial.

En nuestro ejemplo, creamos el archivo api/tasks.py:

api/ <--- o el nombre que le hayas puesto
  - urls.py
  - views.py
  - tasks.py

Dentro definimos una tarea similar a la utilizada en nuestro primer hola mundo:

import time

from celery import shared_task


@shared_task
def saludar(nombre):
    time.sleep(5)

    return f"Saludo enviado a {nombre}"

Volvemos a utilizar time.sleep(5) para simular un trabajo que tarda cinco segundos. En una aplicación real, esta espera sería sustituida por el envío de un correo, la generación de un informe, una llamada a una API externa o cualquier otra operación lenta.

Esta vez no utilizamos:

@app.task

Sino ese @shared_task que acabamos de ver. Esto es importante. Con @app.task necesitaríamos importar directamente la instancia de Celery del proyecto:

from config.celery import app


@app.task
def saludar(nombre):
    ...

Y esto acoplaría nuestra aplicación api al proyecto config. Si quisiéramos reutilizarla dentro de otro proyecto Django, seguiría dependiendo de una instancia concreta de Celery.

@shared_task nos permite registrar la tarea sin importar directamente esa instancia:

from celery import shared_task


@shared_task
def saludar(nombre):
    ...

Cuando Celery descubra api/tasks.py, asociará la tarea con la aplicación de Celery que hemos cargado previamente desde config/__init__.py.

4.2. Publicar la tarea desde una view

Ahora modificamos el endpoint que habíamos preparado en api/views.py:

from django.http import JsonResponse
from django.views.decorators.http import require_GET

from .tasks import saludar


@require_GET
def foo(request):
    nombre = request.GET.get("nombre", "Asdrúbal")
    resultado = saludar.delay(nombre)

    return JsonResponse(
        {
            "task_id": resultado.id,
            "message": "La tarea ha sido enviada",
        },
        status=202,
    )

Obtenemos el nombre desde un parámetro de la URL:

nombre = request.GET.get("nombre", "Asdrúbal")

Si no recibimos ninguno, utilizamos Asdrúbal como valor predeterminado.

La línea importante es:

resultado = saludar.delay(nombre)

Como ya vimos, delay() no ejecuta la función dentro del proceso de Django. Celery crea un mensaje con el nombre de la tarea y sus argumentos y lo publica en Redis:

Tarea: api.tasks.saludar
Argumentos: ["Asdrúbal"]

Django no espera los cinco segundos y responde inmediatamente:

return JsonResponse(
    {
        "task_id": resultado.id,
        "message": "La tarea ha sido enviada",
    },
    status=202,
)

El código de estado HTTP 202 Accepted indica que la petición ha sido aceptada para su procesamiento, pero que el trabajo todavía no tiene por qué haber terminado.

La respuesta tendrá una forma similar a esta:

{
  "task_id": "97dadfc2-ed08-4e60-90fc-c3346f1dc89a",
  "message": "La tarea ha sido enviada"
}

(Para simplificar la prueba estamos utilizando un endpoint GET. En una API real sería más apropiado utilizar POST, porque estamos provocando una operación en el servidor y no limitándonos a consultar información. De momento evitaremos introducir el procesamiento del cuerpo de la petición y la protección CSRF, que no forman parte del objetivo de este tutorial).

4.3. Arrancar el sistema

Para probar la integración necesitamos mantener en ejecución tres procesos diferentes.

Primero comprobamos que Redis está arrancado:

docker start celery-redis

Podemos verificarlo con:

docker exec celery-redis redis-cli ping

La respuesta esperada es:

PONG

En un terminal iniciamos Django:

uv run python manage.py runserver

Y en otro terminal, situado en la misma carpeta que manage.py, iniciamos el worker:

uv run celery -A config worker --loglevel=INFO

El parámetro:

-A config

indica a Celery que debe buscar su aplicación dentro del paquete config.

Celery importa config, ejecuta config/__init__.py y encuentra el objeto que hemos exportado como celery_app:

from .celery import app as celery_app

Al arrancar, el worker debería mostrar nuestra tarea entre las registradas:

[tasks]
  . api.tasks.saludar

Si la tarea no aparece, debemos comprobar principalmente dos cosas:

  • Que api esté incluida en INSTALLED_APPS.
  • Que la tarea se encuentre en un archivo llamado tasks.py.

Ahora podemos llamar al endpoint desde el navegador:

http://localhost:8000/api/foo/?nombre=Aníbal

La ruta exacta dependerá de cómo hayamos configurado urls.py.

Django responderá inmediatamente con el identificador:

{
  "task_id": "ad18d67c-1bc9-47da-a643-c3f041311d71",
  "message": "La tarea ha sido enviada"
}

Mientras tanto, en el terminal del worker veremos algo parecido a esto:

[INFO/MainProcess] Task api.tasks.saludar[ad18d67c-...] received
[INFO/ForkPoolWorker-1] Task api.tasks.saludar[ad18d67c-...] succeeded in 5.00s:
'Saludo enviado a Aníbal'

El recorrido completo que ha hecho es:

  1. Navegador >
  2. Django recibe la petición
  3. saludar.delay("Aníbal")
  4. Redis guarda el mensaje
  5. Django responde con 202 y el task_id
  6. El worker recoge el mensaje
  7. El worker ejecuta la tarea

Django conoce el identificador de la tarea, pero todavía no puede consultar qué ha sucedido con ella. Redis está funcionando únicamente como broker y el resultado mostrado en el terminal del worker no se está almacenando en ningún lugar accesible para Django.

Para poder consultar su estado y recuperar el valor devuelto, el siguiente paso será configurar un backend de resultados.

5. Backend de resultados

5.1. Configuración

Hasta ahora Redis está actuando únicamente como broker:

CELERY_BROKER_URL = "redis://localhost:6379/0"

Django publica una tarea, Redis la mantiene mientras está pendiente y un worker de Celery la recoge y ejecuta. Sin embargo, una vez entregado el mensaje, el broker no conserva el resultado de la ejecución.

Para almacenar el estado y el valor devuelto por las tareas necesitamos configurar un backend de resultados.

Celery admite distintos backends, pero en nuestro ejemplo volveremos a utilizar Redis. Añadimos esta variable a config/settings.py:

CELERY_BROKER_URL = "redis://localhost:6379/0"
CELERY_RESULT_BACKEND = "redis://localhost:6379/1"

Estamos utilizando el mismo servidor Redis, pero dos bases de datos lógicas diferentes:

redis://localhost:6379/0 > broker
redis://localhost:6379/1 > backend de resultados

La base 0 contiene temporalmente los mensajes pendientes de ejecución. La base 1 almacena los estados y resultados de las tareas. Podríamos utilizar la misma base para las dos funciones, pero separarlas nos permite inspeccionar sus contenidos con más claridad y evita mezclar mensajes pendientes con resultados.

5.2. Reiniciar el worker

Después de modificar la configuración debemos reiniciar el worker de Celery. El worker no recarga automáticamente los cambios realizados en settings.py.

Lo detenemos mediante Ctrl + C y volvemos a iniciarlo:

uv run celery -A config worker --loglevel=INFO

En la información de arranque debería aparecer ahora:

transport: redis://localhost:6379/0
results:   redis://localhost:6379/1

Esto nos confirma que Redis desempeña dos funciones distintas:

  • transport indica el broker utilizado para transportar las tareas.
  • results indica el backend utilizado para almacenar sus resultados.

También conviene reiniciar Django para asegurarnos de que utiliza la nueva configuración:

uv run python manage.py runserver

Las tareas ejecutadas antes de configurar el backend no aparecerán mágicamente en él. Solo se almacenarán los resultados de las tareas ejecutadas después del cambio.

5.3. El identificador fundamental

Volvemos a llamar al endpoint:

http://localhost:8000/api/foo/?nombre=Aníbal

Django responderá inmediatamente:

{
  "task_id": "ad18d67c-1bc9-47da-a643-c3f041311d71",
  "message": "La tarea ha sido enviada"
}

El identificador es la clave que nos permitirá consultar posteriormente el estado y el resultado de esa ejecución concreta.

Cuando el worker termine, guardará en el backend:

  • el identificador de la tarea;
  • su estado;
  • el valor devuelto;
  • la fecha de finalización;
  • información sobre posibles errores.

Ahora vamos a crear un endpoint que reciba el identificador de una tarea y consulte su estado.

Modificamos api/views.py:

from celery.result import AsyncResult
from django.http import JsonResponse
from django.views.decorators.http import require_GET

from .tasks import saludar


@require_GET
def foo(request):
    nombre = request.GET.get("nombre", "Asdrúbal")
    resultado = saludar.delay(nombre)

    return JsonResponse(
        {
            "task_id": resultado.id,
            "message": "La tarea ha sido enviada",
        },
        status=202,
    )


@require_GET
def task_status(request, task_id):
    resultado = AsyncResult(task_id)

    response = {
        "task_id": resultado.id,
        "status": resultado.status,
        "ready": resultado.ready(),
    }

    if resultado.successful():
        response["result"] = resultado.result

    return JsonResponse(response)

AsyncResult no ejecuta ninguna tarea. Representa una ejecución concreta a partir de su identificador:

resultado = AsyncResult(task_id)

A través de este objeto podemos consultar el backend:

resultado.status

Devuelve el estado de la tarea.

resultado.ready()

Indica si ha terminado, tanto correctamente como con un error.

resultado.successful()

Indica si terminó correctamente.

resultado.result

Contiene el valor devuelto por la tarea cuando ha terminado con éxito.

Solo incluimos el resultado cuando successful() devuelve True:

if resultado.successful():
    response["result"] = resultado.result

Así evitamos intentar serializar como JSON una excepción cuando la tarea haya fallado. Trataremos esos errores más adelante.

En api/urls.py añadimos la nueva ruta:

from django.urls import path

from . import views


urlpatterns = [
    path("foo/", views.foo, name="foo"),
    path(
        "tasks/<str:task_id>/",
        views.task_status,
        name="task-status",
    ),
]

Y ahora podemos utilizar el identificador devuelto por el primer endpoint:

GET /api/tasks/ad18d67c-1bc9-47da-a643-c3f041311d71/

Si consultamos inmediatamente, mientras la tarea sigue ejecutándose, podemos obtener:

{
  "task_id": "ad18d67c-1bc9-47da-a643-c3f041311d71",
  "status": "PENDING",
  "ready": false
}

Cuando hayan transcurrido los cinco segundos:

{
  "task_id": "ad18d67c-1bc9-47da-a643-c3f041311d71",
  "status": "SUCCESS",
  "ready": true,
  "result": "Saludo enviado a Aníbal"
}

Este endpoint no espera a que termine la tarea. Únicamente consulta la información disponible en ese momento y responde inmediatamente.

Por eso no utilizamos:

resultado.get()

get() esperaría hasta que la tarea terminara y volvería a bloquear el proceso de Django. Con status, ready() y result realizamos una consulta no bloqueante al backend.

Ver el resultado directamente en Redis

También podemos comprobar qué está guardando Redis.

Entramos en la base lógica 1:

docker exec -it celery-redis redis-cli -n 1

Buscamos las claves almacenadas:

SCAN 0 MATCH celery-task-meta-*

Después de completar una tarea aparecerá una clave parecida a esta:

celery-task-meta-ad18d67c-1bc9-47da-a643-c3f041311d71

Celery forma la clave uniendo:

celery-task-meta- + identificador de la tarea

Podemos consultar su contenido con GET:

GET celery-task-meta-ad18d67c-1bc9-47da-a643-c3f041311d71

Obtendremos un documento serializado parecido a este:

{
  "status": "SUCCESS",
  "result": "Saludo enviado a Aníbal",
  "traceback": null,
  "children": [],
  "date_done": "2026-08-30T18:30:00.000000+00:00",
  "task_id": "ad18d67c-1bc9-47da-a643-c3f041311d71"
}

El broker y el backend contienen, por tanto, información completamente diferente:

Redis, base 0
celery
    → mensajes que todavía deben procesarse

Redis, base 1
celery-task-meta-<task_id>
    → estado y resultado de tareas procesadas

Cuando el worker recoge una tarea, su mensaje desaparece de la cola del broker. Cuando termina, el worker escribe su estado y resultado en el backend.

Por defecto, los resultados que se guardan en el Redis caducan después de un día, pero si van a ser muchos resultados hay que tomar medidas especiales, como guardar el archivo en otro sistema y devolver únicamente una referencia.

Bueno, esto último es más complejo, pero de momento vamos a dejarlo aquí.

mmfilesi · 2026
Artículos relacionados