Асинхронные операции в Django с Celery

# Пример настройки Redis как брокера и хранилища результатов
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'

# Пример настройки RabbitMQ
CELERY_BROKER_URL = 'amqp://guest:guest@localhost:5672//'

# myproject/celery.py
import os
from celery import Celery
 
# Устанавливаем переменную окружения для настроек Django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
 
# Создаем экземпляр Celery
app = Celery('myproject')
 
# Загружаем настройки из settings.py (с префиксом CELERY_)
app.config_from_object('django.conf:settings', namespace='CELERY')
 
# Автоматически обнаруживаем задачи в приложениях Django
app.autodiscover_tasks()

# myproject/__init__.py
from .celery import app as celery_app
 
__all__ = ('celery_app',)

# Пример настройки сериализации в settings.py
CELERY_ACCEPT_CONTENT = ['json', 'pickle']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

# Неправильно: передача объекта модели
def process_user_task(user):
    # ...
 
# Правильно: передача идентификатора
def process_user_task(user_id):
    from myapp.models import User
    user = User.objects.get(id=user_id)
    # ...

# Отправка задачи в асинхронном режиме
from myapp.tasks import process_data
result = process_data.delay(arg1, arg2)
# или с дополнительными параметрами
result = process_data.apply_async(args=[arg1, arg2], countdown=10)

# Настройка eager режима в settings.py
CELERY_TASK_ALWAYS_EAGER = True
CELERY_TASK_EAGER_PROPAGATES = True  # Распространение исключений

# Временное включение eager режима для конкретного вызова
with app.conf.eager_propagates_exceptions.override(True):
    with app.conf.task_always_eager.override(True):
        task.delay()

from celery import shared_task
from celery.exceptions import MaxRetriesExceededError
import logging
 
@shared_task(bind=True, max_retries=3, default_retry_delay=60)
def process_payment(self, payment_id):
    try:
        # Код обработки платежа
        payment_gateway.process(payment_id)
    except PaymentGatewayError as exc:
        try:
            # Логирование ошибки
            logging.error(f"Payment error: {exc}")
            # Повторная попытка через 60 секунд
            raise self.retry(exc=exc)
        except MaxRetriesExceededError:
            # Действия после исчерпания попыток
            notify_admin.delay(f"Payment {payment_id} failed after retries")
            raise

@shared_task(bind=True, max_retries=5)
def fetch_api_data(self, endpoint):
    try:
        response = requests.get(endpoint, timeout=5)
        response.raise_for_status()
        return response.json()
    except (requests.RequestException, ValueError) as exc:
        # Экспоненциальная задержка: 2^retry * базовая_задержка
        retry_backoff = 2 ** self.request.retries * 30
        # Случайный "джиттер" для предотвращения "эффекта громады"
        retry_jitter = random.uniform(0, 10)
        retry_delay = retry_backoff + retry_jitter
        raise self.retry(exc=exc, countdown=int(retry_delay))

[program:celery]
command=/path/to/venv/bin/celery -A myproject worker --loglevel=INFO
directory=/path/to/project
user=celery
numprocs=1
stdout_logfile=/var/log/celery/worker.log
stderr_logfile=/var/log/celery/worker.log
autostart=true
autorestart=true
startsecs=10
stopwaitsecs=600

# Ограничения в настройках Celery
CELERY_WORKER_MAX_TASKS_PER_CHILD = 100  # Рестарт воркера после 100 задач
CELERY_WORKER_MAX_MEMORY_PER_CHILD = 200000  # 200MB

# Защита от подделки сообщений
CELERY_BROKER_URL = 'redis://:strong_password@redis.example.com:6379/0'
 
# Ограничение типов сериализации
CELERY_ACCEPT_CONTENT = ['json']  # Отключаем потенциально опасный pickle

# Настройка интеграции с Prometheus
CELERY_WORKER_SEND_TASK_EVENTS = True
CELERY_TASK_SEND_SENT_EVENT = True

[H2]settings.py[/H2]
 
# Настройки Celery
CELERY_BROKER_URL = 'redis://localhost:6379/0'  # URL брокера сообщений
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'  # Хранилище результатов
 
# Дополнительные настройки (опционально)
CELERY_ACCEPT_CONTENT = ['json']  # Разрешенные форматы сериализации
CELERY_TASK_SERIALIZER = 'json'  # Формат сериализации задач
CELERY_RESULT_SERIALIZER = 'json'  # Формат сериализации результатов
CELERY_TIMEZONE = 'Europe/Moscow'  # Временная зона для планировщика задач

# myproject/celery.py
import os
from celery import Celery
 
# Устанавливаем переменную окружения с настройками Django
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
 
# Создаем экземпляр приложения Celery
app = Celery('myproject')
 
# Загружаем настройки из Django settings (с префиксом CELERY_)
app.config_from_object('django.conf:settings', namespace='CELERY')
 
# Автоматически находим задачи в приложениях Django
app.autodiscover_tasks()
 
@app.task(bind=True)
def debug_task(self):
    print(f'Request: {self.request!r}')

# myproject/__init__.py
from __future__ import absolute_import, unicode_literals
 
# Импортируем Celery-приложение
from .celery import app as celery_app
 
__all__ = ('celery_app',)

# users/tasks.py
from celery import shared_task
from django.core.mail import send_mail
from time import sleep
 
@shared_task
def send_welcome_email(user_id):
    from users.models import User  # Импортируем здесь для избежания циклических импортов
    
    try:
        user = User.objects.get(id=user_id)
        # Имитация длительной операции
        sleep(5)
        
        send_mail(
            'Добро пожаловать на наш сайт!',
            f'Здравствуйте, {user.username}! Спасибо за регистрацию.',
            'noreply@example.com',
            [user.email],
            fail_silently=False,
        )
        return f"Email успешно отправлен пользователю {user.email}"
    except User.DoesNotExist:
        return f"Пользователь с ID {user_id} не найден"

# users/views.py
from django.shortcuts import render, redirect
from django.contrib import messages
from .forms import RegistrationForm
from .tasks import send_welcome_email
 
def register(request):
    if request.method == 'POST':
        form = RegistrationForm(request.POST)
        if form.is_valid():
            user = form.save()
            # Запускаем асинхронную задачу
            send_welcome_email.delay(user.id)
            messages.success(request, 'Регистрация успешна! Проверьте вашу почту.')
            return redirect('home')
    else:
        form = RegistrationForm()
    return render(request, 'users/register.html', {'form': form})

# Вариант с apply_async для более гибкой настройки
send_welcome_email.apply_async(
    args=[user.id],
    countdown=10,  # Задержка перед выполнением (сек)
    expires=300,   # Время, после которого задача аннулируется
    retry=True,    # Автоматический повтор при ошибке
    retry_policy={
        'max_retries': 3,
        'interval_start': 0,
        'interval_step': 0.2,
        'interval_max': 0.5,
    }
)

   tasks/
     __init__.py
     emails.py
     reports.py
     cleanup.py
     ...

   # payments/services.py
   from myproject.tasks.registry import register_task
   
   @register_task
   def process_payment(payment_id):
       # Логика обработки платежа

   from celery import chain
   
   # Создание последовательной цепочки задач
   workflow = chain(
       validate_order.s(order_id),
       process_payment.s(),
       send_confirmation.s()
   )
   
   # Запуск цепочки
   result = workflow.delay()

# НЕПРАВИЛЬНО:
@shared_task
def process_article(article):
    article.publish()  # Вызовет ошибку сериализации
 
# ПРАВИЛЬНО:
@shared_task
def process_article(article_id):
    from blog.models import Article
    article = Article.objects.get(id=article_id)
    article.publish()

@shared_task
def notify_subscribers(article_id):
    from blog.models import Article, Subscription
    
    try:
        article = Article.objects.get(id=article_id)
    except Article.DoesNotExist:
        return "Статья больше не существует"
    
    # Используем актуальные данные из БД, а не устаревшие
    subscriptions = Subscription.objects.filter(
        category=article.category,
        is_active=True  # Проверяем статус на момент выполнения
    )
    
    # Остальной код...

# В представлении или сервисе
@transaction.atomic
def create_order(request):
    order = Order.objects.create(user=request.user)
    # Задача может запуститься ДО фиксации транзакции
    process_order.delay(order.id)  # Риск гонки

from django.db import transaction
 
@transaction.atomic
def create_order(request):
    order = Order.objects.create(user=request.user)
    # Задача будет поставлена в очередь ТОЛЬКО после успешной фиксации
    transaction.on_commit(lambda: process_order.delay(order.id))

from celery.result import AsyncResult
 
def check_task_status(request, task_id):
    task = AsyncResult(task_id)
    
    context = {
        'task_id': task_id,
        'status': task.status,
        'result': task.result if task.successful() else None,
    }
    
    return render(request, 'task_status.html', context)

# Для Redis
CELERY_BROKER_URL = 'redis://:сложный_пароль@redis-сервер:6379/0'
 
# Для RabbitMQ
CELERY_BROKER_URL = 'amqp://пользователь:пароль@rabbitmq-сервер:5672/vhost'

CELERY_ACCEPT_CONTENT = ['json']  # Исключаем pickle из соображений безопасности

from celery import chain
 
result = chain(
    extract_data.s(url),  # Первая задача
    transform_data.s(),   # Вторая получит результат первой
    load_data.s(target)   # Третья получит результат второй
)()

from celery import group
 
urls = ['http://example1.com', 'http://example2.com', ...]
result = group(fetch_url.s(url) for url in urls)()

from celery import chord
 
result = chord(
    (process_part.s(data_part) for data_part in partitioned_data),
    aggregate_results.s()  # Эта задача запустится после выполнения всех частей
)()

task.apply_async(
    args=[arg1, arg2],
    link=success_callback.s(),
    link_error=error_callback.s()
)

# tasks.py
@shared_task(bind=True, max_retries=3)
def analyze_dataset(self, dataset_id):
    from analytics.models import Dataset, AnalysisResult
    
    try:
        dataset = Dataset.objects.get(id=dataset_id)
        
        # Имитация ресурсоемких вычислений
        results = {}
        start_time = time.time()
        
        # Допустим, здесь используется pandas, numpy или другая
        # библиотека для анализа данных
        import pandas as pd
        df = pd.DataFrame(dataset.get_data())
        
        # Несколько тяжелых операций
        results['mean'] = df.mean().to_dict()
        results['correlation'] = df.corr().to_dict()
        results['regression'] = perform_regression_analysis(df)
        
        # Сохраняем результаты
        AnalysisResult.objects.create(
            dataset=dataset,
            results=results,
            processing_time=time.time() - start_time
        )
        
        return f"Анализ датасета {dataset_id} завершен успешно"
    except Dataset.DoesNotExist:
        return f"Датасет {dataset_id} не найден"
    except Exception as exc:
        self.retry(exc=exc, countdown=60*5)  # Повторная попытка через 5 минут

@login_required
def analyze_view(request, dataset_id):
    if request.method == 'POST':
        task = analyze_dataset.delay(dataset_id)
        request.session['analysis_task_id'] = task.id
        messages.success(
            request, 
            "Анализ запущен! Вы получите уведомление, когда он завершится."
        )
        return redirect('dataset_detail', dataset_id=dataset_id)
    
    return render(request, 'analytics/analyze_form.html', {
        'dataset_id': dataset_id
    })

@shared_task(rate_limit='100/m')  # Ограничение: 100 писем в минуту
def send_newsletter(campaign_id, recipient_ids=None):
    from marketing.models import Campaign, Subscriber
    
    campaign = Campaign.objects.get(id=campaign_id)
    
    # Если список получателей не указан, берем всех активных подписчиков
    if not recipient_ids:
        recipients = Subscriber.objects.filter(
            is_active=True, 
            subscribed_to_campaigns=True
        )
    else:
        recipients = Subscriber.objects.filter(id__in=recipient_ids)
    
    # Запускаем индивидуальные задачи для каждого получателя
    sent_count = 0
    for recipient in recipients:
        # Используем транзакционную задачу для каждого получателя
        send_personalized_email.delay(
            campaign_id, 
            recipient.id
        )
        sent_count += 1
    
    return f"Запланировано отправок: {sent_count}"
 
@shared_task(bind=True, max_retries=3)
def send_personalized_email(self, campaign_id, subscriber_id):
    from marketing.models import Campaign, Subscriber, EmailLog
    
    try:
        campaign = Campaign.objects.get(id=campaign_id)
        subscriber = Subscriber.objects.get(id=subscriber_id)
        
        # Персонализация шаблона
        context = {
            'first_name': subscriber.first_name,
            'last_name': subscriber.last_name,
            'unsubscribe_url': f"https://example.com/unsubscribe/{subscriber.unsubscribe_token}/"
        }
        
        # Рендеринг HTML и текстовой версии
        html_content = render_to_string(
            'emails/newsletter.html', 
            {**campaign.template_context, **context}
        )
        text_content = strip_tags(html_content)
        
        # Отправка сообщения
        msg = EmailMultiAlternatives(
            campaign.subject,
            text_content,
            settings.DEFAULT_FROM_EMAIL,
            [subscriber.email]
        )
        msg.attach_alternative(html_content, "text/html")
        msg.send()
        
        # Логируем успешную отправку
        EmailLog.objects.create(
            campaign=campaign,
            subscriber=subscriber,
            status='sent'
        )
        
        return f"Email для {subscriber.email} отправлен"
    
    except (Campaign.DoesNotExist, Subscriber.DoesNotExist) as e:
        # Логируем ошибку
        return f"Ошибка: {str(e)}"
    
    except Exception as exc:
        # Если проблема с отправкой, попробуем повторить
        try:
            self.retry(exc=exc, countdown=60*30)  # Повтор через 30 минут
        except self.MaxRetriesExceededError:
            # Логируем окончательную неудачу
            EmailLog.objects.create(
                campaign_id=campaign_id,
                subscriber_id=subscriber_id,
                status='failed',
                error_message=str(exc)
            )
            return f"Не удалось отправить email после нескольких попыток: {str(exc)}"

@shared_task(bind=True, max_retries=5, rate_limit='5/m')
def fetch_currency_rates(self):
    """Получает текущие курсы валют из внешнего API и сохраняет в базу"""
    import requests
    from finance.models import CurrencyRate
    
    API_URL = "https://api.exchangerate-api.com/v4/latest/USD"
    
    try:
        response = requests.get(API_URL, timeout=10)
        response.raise_for_status()  # Вызовет исключение при HTTP-ошибке
        
        data = response.json()
        rates = data.get('rates', {})
        
        # Сохраняем курсы в базу
        batch = []
        timestamp = timezone.now()
        
        for currency, rate in rates.items():
            batch.append(CurrencyRate(
                currency_code=currency,
                rate=Decimal(str(rate)),  # Безопасное преобразование float в Decimal
                source='exchangerate-api',
                timestamp=timestamp
            ))
        
        # Массовое создание записей
        CurrencyRate.objects.bulk_create(batch)
        
        return f"Обновлено {len(batch)} валютных курсов"
    
    except requests.RequestException as exc:
        # Экспоненциальная задержка между повторами
        retry_backoff = 2 ** self.request.retries
        self.retry(exc=exc, countdown=retry_backoff * 60)

@shared_task
def process_uploaded_image(image_id):
    from gallery.models import Image
    
    try:
        img = Image.objects.get(id=image_id)
        original_path = img.file.path
        
        # Создаем миниатюры разных размеров
        from PIL import Image as PILImage
        
        image = PILImage.open(original_path)
        
        # Создаем миниатюру
        thumb_path = original_path.replace('.', '_thumb.')
        image.copy()
        image.thumbnail((200, 200))
        image.save(thumb_path, quality=90)
        
        # Создаем версию среднего размера
        medium_path = original_path.replace('.', '_medium.')
        image.copy()
        image.thumbnail((800, 800))
        image.save(medium_path, quality=90)
        
        # Оптимизируем оригинал (например, сжимаем JPEG)
        if original_path.lower().endswith('.jpg') or original_path.lower().endswith('.jpeg'):
            optimized = PILImage.open(original_path)
            optimized.save(original_path, quality=85, optimize=True)
        
        # Обновляем метаданные в базе
        img.width, img.height = image.size
        img.processed = True
        img.thumbnail_url = thumb_path.replace(settings.MEDIA_ROOT, settings.MEDIA_URL)
        img.medium_url = medium_path.replace(settings.MEDIA_ROOT, settings.MEDIA_URL)
        img.save()
        
        return f"Изображение {image_id} обработано успешно"
    
    except Image.DoesNotExist:
        return f"Изображение {image_id} не найдено"
    except Exception as e:
        # Логируем ошибку и обновляем статус
        img.processed = False
        img.processing_error = str(e)
        img.save()
        raise

# settings.py
from celery.schedules import crontab
 
CELERY_BEAT_SCHEDULE = {
    'ежедневный-отчет': {
        'task': 'reports.tasks.generate_daily_report',
        'schedule': crontab(hour=7, minute=30),  # Каждый день в 7:30
    },
    'обновление-курсов-валют': {
        'task': 'finance.tasks.update_exchange_rates',
        'schedule': 3600.0,  # Каждый час (в секундах)
        'args': (["USD", "EUR", "GBP"],)
    },
    'очистка-сессий': {
        'task': 'core.tasks.clean_expired_sessions',
        'schedule': crontab(hour=3, minute=0, day_of_week='mon'),  # По понедельникам в 3:00
    }
}

[program:celerybeat]
command=/path/to/venv/bin/celery -A myproject beat --loglevel=INFO
directory=/path/to/project
user=celery
numprocs=1
stdout_logfile=/var/log/celery/beat.log
stderr_logfile=/var/log/celery/beat.log
autostart=true
autorestart=true

from django_celery_beat.models import PeriodicTask, IntervalSchedule
 
# Создаем интервал - каждые 10 минут
schedule, created = IntervalSchedule.objects.get_or_create(
    every=10,
    period=IntervalSchedule.MINUTES,
)
 
# Создаем периодическую задачу
PeriodicTask.objects.create(
    interval=schedule,
    name='Синхронизация данных',
    task='data.tasks.sync_external_data',
    args=json.dumps(['arg1', 'arg2']),
    kwargs=json.dumps({
        'param1': 'value1',
    }),
    expires=datetime.utcnow() + timedelta(days=30),
)

@shared_task(
    bind=True,
    max_retries=5,
    default_retry_delay=60,
    autoretry_for=(RequestException, ConnectionError),
    retry_backoff=True,
    retry_backoff_max=600,
    retry_jitter=True
)
def process_payment(self, payment_id):
    try:
        # Основная логика обработки платежа
        payment = Payment.objects.get(id=payment_id)
        
        response = requests.post(
            'https://payment-gateway.example/process',
            json=payment.get_payload(),
            timeout=15
        )
        
        if response.status_code >= 400:
            # Пример специфичной обработки ошибок
            error_data = response.json()
            if error_data.get('error_code') == 'insufficient_funds':
                # Не повторяем, если недостаточно средств
                payment.status = 'failed'
                payment.error_message = 'Недостаточно средств'
                payment.save()
                return False
            elif error_data.get('error_code') == 'gateway_timeout':
                # Специальная задержка для этого типа ошибок
                raise self.retry(countdown=120, exc=Exception("Таймаут шлюза"))
            else:
                # Стандартная обработка других ошибок
                raise self.retry(exc=Exception(f"Ошибка платежного шлюза: {error_data}"))
        
        # Успешная обработка
        payment.status = 'completed'
        payment.save()
        return True
    
    except Payment.DoesNotExist:
        # Логируем и не повторяем
        logger.error(f"Payment {payment_id} not found")
        return False

@shared_task(bind=True)
def process_data(self, data_id):
    try:
        # Основная логика
        return result
    except Exception as exc:
        # Запускаем задачу-обработчик ошибок
        handle_task_error.delay(
            task_id=self.request.id,
            task_name=self.name,
            args=self.request.args,
            kwargs=self.request.kwargs,
            exception=str(exc),
            traceback=traceback.format_exc()
        )
        # Перебрасываем исключение дальше
        raise

# Конфигурация для разных типов воркеров
CELERY_WORKER_POOL_RESTARTS = True  # Автоматический перезапуск пула
CELERY_WORKER_PREFETCH_MULTIPLIER = 1  # Контроль prefetch для баланса нагрузки
 
# Настройки для специализированных очередей
CELERY_TASK_ROUTES = {
    'myapp.tasks.process_video': {'queue': 'media'},
    'myapp.tasks.send_email': {'queue': 'emails'},
    'myapp.tasks.generate_report': {'queue': 'reports'},
}

@shared_task
def optimized_task(large_dataset_id):
    from django.db import connection
    
    # Отключаем автокоммит для пакетных операций
    old_autocommit = connection.autocommit
    connection.autocommit = False
    
    try:
        dataset = LargeDataset.objects.get(id=large_dataset_id)
        items = dataset.items.all()
        
        # Обрабатываем данные пакетами для экономии памяти
        batch_size = 1000
        for i in range(0, items.count(), batch_size):
            batch = items[i:i+batch_size]
            
            # Производим необходимые операции с пакетом
            process_batch(batch)
            
            # Выполняем промежуточный коммит
            connection.commit()
        
        return "Обработка завершена успешно"
    finally:
        # Восстанавливаем исходное значение автокоммита
        connection.autocommit = old_autocommit

def profile_task(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        import cProfile, pstats, io
        from pstats import SortKey
        
        profile = cProfile.Profile()
        profile.enable()
        
        try:
            result = func(*args, **kwargs)
            return result
        finally:
            profile.disable()
            
            # Выводим статистику профилирования
            s = io.StringIO()
            ps = pstats.Stats(profile, stream=s).sort_stats(SortKey.CUMULATIVE)
            ps.print_stats(20)  # Топ-20 самых затратных операций
            
            # Сохраняем результаты
            TaskProfile.objects.create(
                task_name=func.__name__,
                args=str(args),
                kwargs=str(kwargs),
                profile_data=s.getvalue()
            )
    
    return wrapper

# settings/test.py
CELERY_TASK_ALWAYS_EAGER = True
CELERY_TASK_EAGER_PROPAGATES = True  # Пробрасывать исключения

from django.test import TestCase
from myapp.tasks import process_upload
 
class TaskTestCase(TestCase):
    def test_process_upload(self):
        # Задача выполнится синхронно
        result = process_upload.delay(file_id=1)
        self.assertTrue(result.successful())
        self.assertEqual(result.result, "ожидаемый_результат")

from unittest.mock import patch
from django.test import TestCase
from myapp.tasks import send_notification
from myapp.models import User
 
class NotificationTaskTest(TestCase):
    def setUp(self):
        self.user = User.objects.create(email="test@example.com")
    
    @patch('myapp.tasks.send_mail')
    def test_send_notification(self, mock_send_mail):
        # Вызываем задачу напрямую
        send_notification(self.user.id)
        
        # Проверяем, что send_mail был вызван с правильными аргументами
        mock_send_mail.assert_called_once()
        args, kwargs = mock_send_mail.call_args
        self.assertEqual(kwargs['recipient_list'], [self.user.email])

import time
from django.test import TransactionTestCase
from celery.result import AsyncResult
from myapp.tasks import complex_task
 
class CeleryIntegrationTest(TransactionTestCase):
    def test_complex_task_integration(self):
        # Запускаем реальную асинхронную задачу
        task = complex_task.delay(param1="test", param2=123)
        
        # Дожидаемся результата с таймаутом
        timeout = 10
        start_time = time.time()
        while time.time() - start_time < timeout:
            if AsyncResult(task.id).ready():
                break
            time.sleep(0.1)
        
        # Проверяем результат
        result = AsyncResult(task.id)
        self.assertTrue(result.ready())
        self.assertTrue(result.successful())
        self.assertEqual(result.result, "ожидаемый_результат")

# settings.py
CELERY_TASK_ROUTES = {
    'myapp.tasks.critical_notification': {'queue': 'high_priority'},
    'myapp.tasks.daily_report': {'queue': 'low_priority'},
    'myapp.tasks.*': {'queue': 'default'},
}

class TaskPriority:
    HIGH = 0
    MEDIUM = 3
    LOW = 6
    BACKGROUND = 9
 
@shared_task(bind=True)
def prioritized_task(self, importance, data):
    # Логика задачи
    pass
 
# Вызов с указанием приоритета
prioritized_task.apply_async(
    args=[TaskPriority.HIGH, data],
    priority=TaskPriority.HIGH
)

@shared_task(bind=True)
def instrumented_task(self, *args, **kwargs):
    import time
    from prometheus_client import Histogram
    
    TASK_DURATION = Histogram(
        'task_duration_seconds',
        'Task duration in seconds',
        ['task_name']
    )
    
    start_time = time.time()
    try:
        result = actual_task_logic(*args, **kwargs)
        status = 'success'
    except Exception as e:
        status = 'error'
        raise
    finally:
        duration = time.time() - start_time
        TASK_DURATION.labels(self.name).observe(duration)
        
        # Сохраняем детальную информацию в БД
        TaskExecution.objects.create(
            task_id=self.request.id,
            task_name=self.name,
            args=str(args),
            kwargs=str(kwargs),
            duration=duration,
            status=status,
            worker=self.request.hostname,
            memory_usage=get_current_memory_usage()
        )
    
    return result

def task_performance_dashboard(request):
    # Агрегированная статистика по задачам
    task_stats = TaskExecution.objects.values('task_name').annotate(
        avg_duration=Avg('duration'),
        max_duration=Max('duration'),
        min_duration=Min('duration'),
        success_rate=100 * Count('id', filter=Q(status='success')) / Count('id'),
        count=Count('id')
    ).order_by('-avg_duration')
    
    # График выполнения по времени
    timeline_data = TaskExecution.objects.filter(
        created_at__gte=timezone.now() - timezone.timedelta(days=7)
    ).values('created_at__date').annotate(
        count=Count('id'),
        success=Count('id', filter=Q(status='success')),
        failed=Count('id', filter=Q(status='error'))
    ).order_by('created_at__date')
    
    return render(request, 'tasks/dashboard.html', {
        'task_stats': task_stats,
        'timeline_data': timeline_data
    })

# Тестирование новой версии параллельно с текущей
from celery import Celery
 
# Существующее приложение
current_app = Celery('current_app', broker='redis://localhost:6379/0')
 
# Тестовое приложение с новой версией
new_app = Celery('new_app', broker='redis://localhost:6379/1')
 
@current_app.task
def current_version_task():
    pass
 
@new_app.task
def new_version_task():
    pass
 
# Запуск задачи в обеих версиях для сравнения
def compare_versions():
    current_result = current_version_task.delay()
    new_result = new_version_task.delay()
    # Сравнение результатов и поведения