8e69178e07
Instead of inserting the path of the apps into the Python app, the apps are now referenced by their full import path. This app name claves with external or native Python libraries. Example: Mayan statistics app vs. Python new statistics library. Every app reference is now prepended with 'mayan.apps'. Existing config.yml files need to be updated manually. Signed-off-by: Roberto Rosario <roberto.rosario.gonzalez@gmail.com>
72 lines
2.0 KiB
Python
72 lines
2.0 KiB
Python
from __future__ import absolute_import, unicode_literals
|
|
|
|
from django.contrib import messages
|
|
from django.utils.translation import ugettext_lazy as _
|
|
|
|
from mayan.apps.common.generics import SingleObjectListView
|
|
|
|
from .classes import CeleryQueue
|
|
from .permissions import permission_task_view
|
|
|
|
|
|
class QueueListView(SingleObjectListView):
|
|
extra_context = {
|
|
'hide_object': True,
|
|
'title': _('Background task queues'),
|
|
}
|
|
view_permission = permission_task_view
|
|
|
|
def get_object_list(self):
|
|
return CeleryQueue.all()
|
|
|
|
|
|
class QueueActiveTaskListView(SingleObjectListView):
|
|
view_permission = permission_task_view
|
|
|
|
def get_extra_context(self):
|
|
return {
|
|
'hide_object': True,
|
|
'object': self.get_object(),
|
|
'title': _('Active tasks in queue: %s') % self.get_object()
|
|
}
|
|
|
|
def get_object(self):
|
|
return CeleryQueue.get(queue_name=self.kwargs['queue_name'])
|
|
|
|
def get_object_list(self):
|
|
try:
|
|
return self.get_task_list()
|
|
except Exception as exception:
|
|
messages.error(
|
|
self.request,
|
|
_('Unable to retrieve task list; %s') % exception
|
|
)
|
|
return ()
|
|
|
|
def get_task_list(self):
|
|
return self.get_object().get_active_tasks()
|
|
|
|
|
|
class QueueScheduledTaskListView(QueueActiveTaskListView):
|
|
def get_extra_context(self):
|
|
return {
|
|
'hide_object': True,
|
|
'object': self.get_object(),
|
|
'title': _('Scheduled tasks in queue: %s') % self.get_object()
|
|
}
|
|
|
|
def get_task_list(self):
|
|
return self.get_object().get_scheduled_tasks()
|
|
|
|
|
|
class QueueReservedTaskListView(QueueActiveTaskListView):
|
|
def get_extra_context(self):
|
|
return {
|
|
'hide_object': True,
|
|
'object': self.get_object(),
|
|
'title': _('Reserved tasks in queue: %s') % self.get_object()
|
|
}
|
|
|
|
def get_task_list(self):
|
|
return self.get_object().get_reserved_tasks()
|