Improve backup admin error visibility and log celery tracebacks
Some checks failed
CI/CD Pipeline / Code Quality Checks (push) Failing after 2m9s
CI/CD Pipeline / Run Tests (push) Failing after 2m28s
CI/CD Pipeline / Run API Inventory E2E Tests (push) Has been skipped
CI/CD Pipeline / Telegram Notify Success (push) Has been skipped

This commit is contained in:
2026-04-22 10:51:00 +02:00
parent 92d5ff4252
commit c26df87d80
2 changed files with 107 additions and 1 deletions

View File

@@ -3,12 +3,15 @@
from apps.backups.models import BackupExportJob
from apps.backups.serializers import BackupExportRequestSerializer
from apps.backups.services import BackupExportError, BackupExportJobService
from apps.core.services import BackgroundJobService
from django.contrib import admin
from django.contrib import messages
from django.http import HttpResponse
from django.shortcuts import redirect
from django.urls import path, reverse
from django.utils.html import format_html
from django.utils import timezone
from urllib.parse import urlencode
@admin.register(BackupExportJob)
@@ -23,9 +26,11 @@ class BackupExportJobAdmin(admin.ModelAdmin):
"requested_by",
"organizations_count",
"archive_size",
"error_short",
"started_at",
"completed_at",
]
list_display_links = ["id"]
list_filter = ["status", "actual_date", "created_at", "completed_at"]
search_fields = ["task_id", "checksum_sha256", "archive_filename"]
readonly_fields = [
@@ -37,14 +42,113 @@ class BackupExportJobAdmin(admin.ModelAdmin):
"archive_size",
"organizations_count",
"error",
"background_job_link",
"background_job_error",
"background_job_traceback",
"started_at",
"completed_at",
"created_at",
"updated_at",
]
fieldsets = (
(
"Основные поля",
{"fields": ("actual_date", "status", "task_id", "background_job_link")},
),
(
"Исполнитель",
{"fields": ("requested_by",)},
),
(
"Результат",
{
"fields": (
"archive_path",
"archive_filename",
"checksum_filename",
"checksum_sha256",
"archive_size",
"organizations_count",
)
},
),
(
"Ошибка",
{
"fields": (
"error",
"background_job_error",
"background_job_traceback",
),
"classes": ("collapse",),
},
),
(
"Время",
{"fields": ("started_at", "completed_at", "created_at", "updated_at")},
),
)
ordering = ["-actual_date", "-created_at"]
change_list_template = "admin/backups/backupexportjob/change_list.html"
@staticmethod
def _format_error_text(text: str) -> str:
"""Сократить и безопасно форматировать текст ошибки для списка."""
if not text:
return ""
text = text.strip()
return text if len(text) <= 120 else f"{text[:117]}"
def error_short(self, obj: BackupExportJob) -> str:
"""Короткий текст ошибки для списка."""
return self._format_error_text(obj.error)
error_short.short_description = "Ошибка"
def _get_background_job(self, obj: BackupExportJob):
if not obj.task_id:
return None
return BackgroundJobService.get_by_task_id_or_none(obj.task_id)
def background_job_link(self, obj: BackupExportJob):
background_job = self._get_background_job(obj)
if not background_job:
return ""
changelist_url = (
reverse("admin:core_backgroundjob_changelist")
+ "?"
+ urlencode({"q": background_job.task_id})
)
return format_html(
'<a href="{}">Открыть в журнале задач: {}</a>',
changelist_url,
background_job.task_id,
)
background_job_link.short_description = "Celery задача"
def background_job_error(self, obj: BackupExportJob) -> str:
background_job = self._get_background_job(obj)
if not background_job:
return ""
if background_job.error:
return background_job.error
return ""
background_job_error.short_description = "Ошибка в Celery"
def background_job_traceback(self, obj: BackupExportJob):
background_job = self._get_background_job(obj)
if not background_job or not background_job.traceback:
return ""
return format_html(
'<pre style="white-space: pre-wrap; max-height: 400px; overflow: auto; '
'background: #f8f9fa; border: 1px solid #dee2e6; padding: 8px;">{}</pre>',
background_job.traceback,
)
background_job_traceback.short_description = "Traceback из фоновой задачи"
def get_urls(self):
urls = super().get_urls()
custom_urls = [

View File

@@ -3,6 +3,7 @@
from __future__ import annotations
import logging
import traceback
import uuid
from pathlib import Path
@@ -109,9 +110,10 @@ def generate_backup_for_date(self, job_id: int) -> dict:
job.actual_date,
job.id,
)
error_traceback = traceback.format_exc()
job.status = BackupExportJob.Status.FAILURE
job.error = str(exc)
job.completed_at = timezone.now()
job.save(update_fields=["status", "error", "completed_at", "updated_at"])
background_job.fail(error=str(exc))
background_job.fail(error=str(exc), traceback_str=error_traceback)
raise