129 lines
3.8 KiB
Python
129 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",
|
|
"app",
|
|
"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']
|