Referência da API

django_celery_task_monitor.models

class django_celery_task_monitor.models.TaskLog(*args, **kwargs)

Registra a execução de uma tarefa Celery vinculada a um objeto qualquer.

Instâncias são tipicamente criadas manualmente pela view/admin do projeto host logo após chamar minha_task.delay(...), informando o content_type/object_id do objeto relacionado e o task_id retornado pelo Celery.

exception DoesNotExist

exception MultipleObjectsReturned

exception NotUpdated

as_status_payload(user) → dict[str, Any]

Serializa o estado atual da tarefa para o endpoint REST de polling.

Além do status “cru”, inclui started_at (quando a execução realmente começou, segundo o TaskResult), progress (dict de progresso, se a task publicou um) e message (frase pronta, equivalente à de get_status_message()) — o JavaScript do plugin recompõe essa frase a cada segundo no cliente (para o relógio de tempo decorrido andar entre um poll e outro), mas message já serve pronta para qualquer outro consumidor do endpoint.

content_object

content_type

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

content_type_id

created_at

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

get_error_details(user) → dict[str, Any]

Retorna os detalhes de erro da tarefa, respeitando a permissão do usuário.

Superusuários e usuários com a permissão view_task_trace recebem o stacktrace completo em traceback. Demais usuários recebem traceback=None e apenas a mensagem amigável configurada em django_celery_task_monitor.settings.FRIENDLY_ERROR_MESSAGE.

get_next_by_created_at(*, field=<django.db.models.fields.DateTimeField: created_at>, is_next=True, **kwargs)

get_next_by_updated_at(*, field=<django.db.models.fields.DateTimeField: updated_at>, is_next=True, **kwargs)

get_previous_by_created_at(*, field=<django.db.models.fields.DateTimeField: created_at>, is_next=False, **kwargs)

get_previous_by_updated_at(*, field=<django.db.models.fields.DateTimeField: updated_at>, is_next=False, **kwargs)

get_progress(task_result=None) → dict[str, Any] | None

Retorna o dict de progresso publicado por um estado customizado.

Quando uma task chama self.update_state(state="PROGRESS", meta={"percent": 42}), o backend de resultados grava meta no campo result do TaskResult (serializado, normalmente como JSON). Este método decodifica esse valor e o retorna como dict — ou None para estados padrão do Celery (PENDING, STARTED, etc., que não carregam progresso) ou quando o valor não é um JSON/objeto decodificável.

get_status_display(*, field=<django.db.models.fields.CharField: status>)

get_status_message() → str

Frase de status legível (ex.: “Tarefa em processamento há 12s.”).

Usada para o render inicial do painel ao vivo (get_urls()’s template task_status_panel.html), antes do JavaScript assumir via polling. Note que, sem CELERY_TASK_TRACK_STARTED = True no projeto host, o Celery não registra quando a execução começou de fato — nesse caso a tarefa aparenta ficar “enfileirada” até o resultado final, mesmo já estando em execução (ver docs/configuration.rst).

get_traceback() → str | None

Retorna o stacktrace bruto do TaskResult, sem verificação de permissão.

Uso interno / para chamadores que já validaram a permissão do usuário (ex.: um campo do admin já removido da tela para quem não tem acesso). Prefira get_error_details() ao expor dados diretamente a um usuário.

id

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

property is_finished: bool

True quando o status atual é terminal (sucesso, falha ou revogada).

object_id

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

objects = <django.db.models.manager.Manager object>

started_by

Accessor to the related object on the forward side of a many-to-one or one-to-one (via ForwardOneToOneDescriptor subclass) relation.

In the example:

class Child(Model):
    parent = ForeignKey(Parent, related_name='children')

Child.parent is a ForwardManyToOneDescriptor instance.

started_by_id

status

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

task_id

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

task_name

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

update_status(save: bool = True, task_result=None) → str

Sincroniza status com o TaskResult mais recente da tarefa.

Se o TaskResult ainda não existir (a tarefa não terminou ou o backend de resultados ainda não persistiu), o status atual é mantido. Retorna o status resultante (novo ou inalterado). Aceita um task_result já carregado para evitar uma consulta redundante (ver as_status_payload()).

updated_at

A wrapper for a deferred-loading field. When the value is read from this object the first time, the query is executed.

Estados possíveis (TaskLog.status): PENDING, STARTED, RETRY, SUCCESS, FAILURE, REVOKED — os mesmos nomes usados pelo Celery (celery.states), espelhados localmente para que este módulo não precise importar o pacote celery diretamente. PROGRESS também é reconhecido (estado customizado comum para progresso percentual, ver Uso Avançado), mas qualquer outro nome de estado customizado também é aceito — o campo status não é restrito aos valores acima.

django_celery_task_monitor.admin

class django_celery_task_monitor.admin.CeleryTaskMonitorMixin(*args, **kwargs)

Mixin que adiciona uma coluna de status de tarefa Celery a um ModelAdmin.

Uso mínimo:

@admin.register(MeuModelo)
class MeuModeloAdmin(CeleryTaskMonitorMixin, admin.ModelAdmin):
    list_display = ["nome", "task_status_column"]

Atributos configuráveis na subclasse:

• celery_poll_interval: intervalo de polling em milissegundos, sobrescrevendo o default global CELERY_TASK_MONITOR_POLL_INTERVAL.

• celery_task_field: nome do atributo/coluna usado no list_display (default: "task_status_column"). Útil quando o nome padrão colide com outro atributo já existente na ModelAdmin.

celery_poll_interval: int | None = None

celery_task_field: str = 'task_status_column'

create_task_log(request: HttpRequest, obj, task_id: str, task_name: str = '') → TaskLog

Registra um TaskLog para uma tarefa já disparada.

Uso de baixo nível — para o caso comum (disparar a tarefa e registrar em seguida), prefira start_task(). Use este método diretamente quando a tarefa já foi disparada de outro jeito (ex.: apply_async() com opções customizadas, ou uma chamada em outro lugar do código que só te devolveu o task_id).

Funciona a partir de qualquer lugar do ModelAdmin — não depende de nenhum hook específico do Django admin (serve para response_change, para uma action de changelist chamada uma vez por objeto do queryset, ou para uma view customizada).

get_celery_poll_interval() → int

Retorna o intervalo de polling (ms) efetivo desta ModelAdmin.

get_urls()

Registra a rota REST de polling antes das rotas padrão do admin.

property media: Media

Garante que task-poll.js seja carregado no changelist/changeform.

task-poll.js se auto-inicializa em qualquer elemento com data-poll-url (ver o próprio arquivo), então nenhum código extra é necessário para o badge começar a fazer polling assim que a página carrega.

render_change_form(request, context, add=False, change=False, form_url='', obj=None)

Injeta task_log_panel_html no contexto do change form.

Diferente do badge do changelist (adicionado automaticamente via list_display), o painel do change form é opt-in: o template change_form.html do seu ModelAdmin precisa referenciar {{ task_log_panel_html }} onde quiser exibi-lo (ex.: logo após o botão que dispara a tarefa). Isso evita que o plugin sobrescreva globalmente o template de todo ModelAdmin do projeto host.

start_task(request: HttpRequest, obj, task, *args, task_name: str | None = None, **kwargs) → TaskLog

Dispara task.delay(*args, **kwargs) e já registra o TaskLog.

Substitui o boilerplate repetido em toda ModelAdmin que dispara tarefas Celery:

task = minha_task.delay(obj.id)
TaskLog.objects.create(
    content_type=ContentType.objects.get_for_model(obj),
    object_id=obj.id,
    task_id=task.id,
    task_name="minha_task",
    started_by=request.user,
)

vira:

self.start_task(request, obj, minha_task, obj.id)

task_name é derivado automaticamente de task.name (todo @shared_task/@app.task tem esse atributo) — só passe task_name= explicitamente para sobrescrever. Funciona igual em response_change, em uma action de changelist (uma chamada por objeto do queryset) ou em qualquer outro lugar do ModelAdmin, sem exigir nenhum hook específico do admin. Retorna o TaskLog recém-criado (já com task_id preenchido).

task_status_column(obj)

Coluna padrão de status. Adicione "task_status_column" ao list_display.

task_status_view(request: HttpRequest, task_id: str) → JsonResponse

Endpoint REST de polling: retorna o status (JSON) de uma tarefa.

Exige permissão de visualização do modelo administrado (a mesma permissão usada para acessar o changelist).

class django_celery_task_monitor.admin.TaskLogAdmin(model, admin_site)

Interface de administração para consultar todos os registros de tarefas.

Somente leitura: instâncias de TaskLog são sempre criadas programaticamente pelo projeto host (ver README.md), então esta ModelAdmin desabilita a criação manual.

date_hierarchy = 'created_at'

friendly_message(obj: TaskLog) → str

Mensagem de erro amigável, visível a qualquer usuário com acesso à tarefa.

full_traceback(obj: TaskLog) → str

Stacktrace completo.

Este campo já é removido em get_fields() para usuários sem a permissão view_task_trace, então, quando chamado, o acesso já foi validado — não é necessário reverificar a permissão aqui.

get_fields(request: HttpRequest, obj: TaskLog | None = None)

Remove full_traceback dos campos exibidos a quem não tem permissão.

has_add_permission(request: HttpRequest) → bool

TaskLog é sempre criado ao disparar a tarefa, nunca manualmente.

list_display = ('task_id', 'task_name', 'content_type', 'object_id', 'status_badge', 'started_by', 'created_at', 'updated_at')

list_filter = ('status', 'task_name', 'content_type')

list_per_page = 50

property media

readonly_fields = ('content_type', 'object_id', 'task_id', 'task_name', 'status', 'started_by', 'created_at', 'updated_at', 'friendly_message', 'full_traceback')

search_fields = ('task_id', 'task_name', 'object_id')

status_badge(obj: TaskLog) → str

Renderiza o badge de status reutilizando o template do plugin.

django_celery_task_monitor.views

class django_celery_task_monitor.views.TaskStatusView(**kwargs)

View genérica que retorna o status (JSON) de um TaskLog pelo task_id.

Para usar, registre a rota no urls.py do projeto host:

from django_celery_task_monitor.views import TaskStatusView

urlpatterns = [
    path("task-status/<str:task_id>/", TaskStatusView.as_view(), name="task-status"),
]

get(request: HttpRequest, task_id: str) → JsonResponse

Retorna o payload de status da tarefa, ou 404 se task_id não existir.

raise_exception = False

django_celery_task_monitor.permissions

django_celery_task_monitor.permissions.user_can_view_task_trace(user: Any) → bool

Retorna True se user pode ver o stacktrace completo de uma tarefa.

Superusuários sempre podem. Usuários anônimos (ou None) nunca podem. Demais usuários precisam da permissão configurada em django_celery_task_monitor.settings.TASK_TRACE_PERMISSION.

user aceita tanto um AbstractBaseUser autenticado quanto um AnonymousUser/None, por isso o tipo é deliberadamente duck-typed (has_perm existe em ambos, mas com bases distintas no Django).

django_celery_task_monitor.settings

Ver Configuração para a lista completa de settings lidas deste módulo.

Template tags ({% load task_monitor_tags %})

Tag	Descrição
{% task_status_badge task_log %}	Renderiza o badge de status (rótulo curto) de um TaskLog.
{% task_status_panel task_log %}	Renderiza o painel de status ao vivo (frase completa) de um TaskLog.
{% task_poll_script selector %}	Emite o <script> do plugin já com a inicialização do polling para selector.
{% task_monitor_static_url %}	URL estática de task-poll.js.
{% task_monitor_static_css_url %}	URL estática do CSS opcional do badge.

Templates

Template	Uso
django_celery_task_monitor/task_status_badge.html	Badge de status usado no changelist, no TaskLogAdmin e na tag task_status_badge.
django_celery_task_monitor/task_status_panel.html	Painel de status ao vivo, injetado no change form por CeleryTaskMonitorMixin.render_change_form() e usado pela tag task_status_panel.
django_celery_task_monitor/task_log_detail.html	Bloco de detalhe de um TaskLog (status, metadados, erro), para uso livre em templates do projeto host.