(sender, task_id, **kwargs)
| 50 | |
| 51 | @receiver(task_failed) |
| 52 | def handle_task_failed(sender, task_id, **kwargs): |
| 53 | if get_current_plugin(only_active=True) is None: |
| 54 | return |
| 55 | |
| 56 | logger.info("TaskNotification: Task Failed") |
| 57 | |
| 58 | config_data = config.load() |
| 59 | if config_data.get("notify_task_failed") == True: |
| 60 | task = Task.objects.get(id=task_id) |
| 61 | setting = Setting.objects.first() |
| 62 | notification_app_name = config_data['notification_app_name'] or settings.app_name |
| 63 | console_output = reverse_output(task.console.output()) |
| 64 | notification.send( |
| 65 | f"{notification_app_name} - {task.project.name} Task Failed", |
| 66 | f"{task.project.name}\n{task.name} Failed with error: {task.last_error}\nProcessing time:{hours_minutes_secs(task.processing_time)}\n\nConsole Output:{console_output}", |
| 67 | config_data |
| 68 | ) |
| 69 | |
| 70 | def hours_minutes_secs(milliseconds): |
| 71 | if milliseconds == 0 or milliseconds == -1: |
nothing calls this directly
no test coverage detected