Anna Sirota
707c283905
This allows identifying failed and completed webhooks more easily, because tasks initiated by webhooks set their verbose names to the names of webhooks.
128 lines
3.8 KiB
Python
128 lines
3.8 KiB
Python
import json
|
|
from typing import Type
|
|
from functools import lru_cache
|
|
|
|
from django.contrib import admin
|
|
from django.contrib.admin import ModelAdmin
|
|
from django.urls import reverse
|
|
from django.utils.html import format_html
|
|
import background_task.admin
|
|
import background_task.models
|
|
import django.db.models
|
|
|
|
from . import models
|
|
|
|
from bid_main.admin_decorators import short_description
|
|
|
|
|
|
@lru_cache(maxsize=100)
|
|
def _load_task_params(task_params_str):
|
|
return json.loads(task_params_str)
|
|
|
|
|
|
@lru_cache(maxsize=100)
|
|
def _get_task_webhook(webhook_pk):
|
|
return models.Webhook.objects.get(pk=webhook_pk)
|
|
|
|
|
|
def _get_admin_url_name(model: Type['django.db.models.Model'], action: str = 'change') -> str:
|
|
return 'admin:{}_{}_{}'.format(model._meta.app_label, model._meta.model_name, action)
|
|
|
|
|
|
def get_admin_change_url(model: Type['django.db.models.Model'], pk: int) -> str:
|
|
"""Return an admin URL to admin change view for the given object."""
|
|
return reverse(_get_admin_url_name(model), args=[pk])
|
|
|
|
|
|
@short_description("Disable selected webhooks")
|
|
def disable_webhook(modeladmin, request, queryset):
|
|
queryset.update(enabled=False)
|
|
|
|
|
|
@short_description("Enable selected webhooks")
|
|
def enable_webhook(modeladmin, request, queryset):
|
|
queryset.update(enabled=True)
|
|
|
|
|
|
@admin.register(models.Webhook)
|
|
class WebhookAdmin(ModelAdmin):
|
|
list_display = (
|
|
"name",
|
|
"hook_type",
|
|
"url",
|
|
"enabled",
|
|
)
|
|
list_display_links = ("name", "hook_type", "url")
|
|
list_filter = ("hook_type", "enabled")
|
|
search_fields = ("name", "hook_type", "url")
|
|
ordering = ("name",)
|
|
actions = [disable_webhook, enable_webhook]
|
|
|
|
|
|
try:
|
|
admin.site.unregister(background_task.models.Task)
|
|
admin.site.unregister(background_task.models.CompletedTask)
|
|
except admin.site.NotRegistered:
|
|
pass
|
|
|
|
|
|
class TaskWebhookMixin:
|
|
"""Modify a few properties of background tasks displayed in admin."""
|
|
|
|
def no_errors(self, obj):
|
|
"""Replace background_task's "has_error".
|
|
|
|
Make Django's red/green boolean icons less confusing
|
|
in the context of "there's an error during task run".
|
|
"""
|
|
return not bool(obj.last_error)
|
|
|
|
no_errors.boolean = True
|
|
|
|
def webhook(self, obj):
|
|
"""Extract which webhook initiated this task, if applicable.
|
|
|
|
TODO(anna): replace with obj.creator after it's set for all tasks.
|
|
"""
|
|
if obj.task_name != 'bid_api.tasks.webhook_send':
|
|
return '-'
|
|
params = _load_task_params(obj.task_params)
|
|
webhook_pk = params[0][0]
|
|
link = get_admin_change_url(models.Webhook, webhook_pk)
|
|
webhook = _get_task_webhook(webhook_pk)
|
|
return format_html('<a target="_blank" href="{}">{}</a>', link, webhook)
|
|
|
|
|
|
@admin.register(background_task.models.Task)
|
|
class TaskAdmin(background_task.admin.TaskAdmin, TaskWebhookMixin):
|
|
date_hierarchy = 'run_at'
|
|
list_filter = ('task_name', 'run_at', 'failed_at', 'locked_at', 'attempts', 'verbose_name')
|
|
search_fields = ['task_name', 'task_params', 'last_error', 'verbose_name']
|
|
list_display = [
|
|
'run_at',
|
|
'task_name',
|
|
'webhook',
|
|
'task_params',
|
|
'attempts',
|
|
'no_errors',
|
|
'locked_by',
|
|
'locked_by_pid_running',
|
|
]
|
|
readonly_fields = ['webhook']
|
|
|
|
|
|
@admin.register(background_task.models.CompletedTask)
|
|
class CompletedTaskAdmin(background_task.admin.CompletedTaskAdmin, TaskWebhookMixin):
|
|
date_hierarchy = 'run_at'
|
|
search_fields = ['task_name', 'task_params', 'last_error', 'verbose_name']
|
|
list_display = [
|
|
'run_at',
|
|
'task_name',
|
|
'webhook',
|
|
'task_params',
|
|
'attempts',
|
|
'no_errors',
|
|
]
|
|
list_filter = ('task_name', 'run_at', 'failed_at', 'locked_at', 'attempts', 'verbose_name')
|
|
readonly_fields = ['webhook']
|