-
Mattia Mancini authoredMattia Mancini authored
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
views.py 17.74 KiB
import logging
from . import config
from django.contrib.auth.decorators import login_required
from django.views.generic import ListView
from django.contrib import messages
from rest_framework import generics, pagination
from rest_framework.response import Response
from django_filters import rest_framework as filters
from django_filters.views import FilterView
from django_tables2.views import SingleTableMixin
from django.shortcuts import render, redirect
from django.core.paginator import Paginator, EmptyPage, PageNotAnInteger
from django_tables2 import SingleTableView
from django.conf import settings
from .models import Task, Status, Workflow, LogEntry, Configuration, Job
from .tables import TaskTable
from django.db.models import Q
from .serializers import \
TaskWriteSerializer, \
TaskReadSerializer, \
TaskReadSerializerFast, \
WorkflowSerializer,\
LogEntrySerializer,\
ConfigurationSerializer,\
JobSerializer
from .services import algorithms
logger = logging.getLogger(__name__)
# ---------- filters (in the REST API) ---------
class TaskFilter(filters.FilterSet):
class Meta:
model = Task
fields = {
'creationTime': ['icontains'],
'filter': ['exact', 'icontains'],
'workflow__id' : ['exact', 'icontains'],
'project': ['exact', 'icontains'],
'sas_id': ['exact', 'icontains'],
'status': ['exact', 'icontains', 'in', 'startswith'],
'purge_policy': ['exact'],
'priority': ['exact','lte','gte'],
'resume': ['exact'],
}
class TaskFilterQueryPage(filters.FilterSet):
class Meta:
model = Task
fields = {
'id': ['exact', 'gte', 'lte'],
'workflow__id': ['exact'],
'filter': ['exact', 'icontains'],
'priority': ['exact', 'gte', 'lte'],
'status': ['icontains', 'in'],
'project': ['exact', 'icontains'],
'sas_id': ['exact', 'icontains'],
}
class WorkflowFilter(filters.FilterSet):
class Meta:
model = Workflow
fields = {
'repository': ['exact', 'icontains'],
'commit_id': ['exact', 'icontains'],
'path': ['exact', 'icontains'],
}
class LogEntryFilter(filters.FilterSet):
class Meta:
model = LogEntry
fields = {
'task__id': ['exact'],
'step_name': ['exact', 'icontains', 'in', 'startswith'],
'status': ['exact','in'],
}
class ConfigurationFilter(filters.FilterSet):
class Meta:
model = Configuration
fields = {
'filter': ['exact', 'icontains'],
'key': ['exact', 'icontains'],
'value': ['exact', 'icontains'],
}
class JobFilter(filters.FilterSet):
class Meta:
model = Job
fields = {
'type': ['exact', 'icontains'],
'task_id': ['exact'],
'job_id': ['exact'],
}
# ---------- Tables2 Views (experimental) -----------
# implementation with tables2: http://localhost:8000/atdb/tables2
class QueryView(SingleTableMixin, FilterView):
table_class = TaskTable
model = Task
template_name = "query/index.html"
filterset_class = TaskFilterQueryPage
def get_table_data(self):
# https://stackoverflow.com/questions/7763115/django-passing-data-between-views
count = self.object_list.count()
try:
limit = int(Configuration.objects.get(key='multi_change_limit').value)
except:
limit = settings.QUERY_LIMIT_MULTI_CHANGE
query_list_of_ids = list(self.object_list.values_list('id'))[:limit]
# store on the session
self.request.session['query_list_of_ids'] = query_list_of_ids
return self.object_list
# ---------- GUI Views -----------
class IndexView(ListView):
"""
This is the main view of ATDB. It shows a pagination list of tasks, sorted by creationTime.
"""
template_name = 'taskdatabase/index.html'
# by default this returns the list in an object called object_list, so use 'object_list' in the html page.
# but if 'context_object_name' is defined, then this returned list is named and can be accessed that way in html.
context_object_name = 'my_tasks'
def get_queryset(self):
status = self.request.GET.get('status')
search_box = self.request.GET.get('search_box', None)
# get the sort variable from the session or use default
try:
sort = self.request.session['sort']
except:
sort = '-creationTime'
tasks = Task.objects.order_by(sort)
# check if there is a 'task_filter' put on the session
try:
filter = self.request.session['task_filter']
if filter!='all':
tasks = get_searched_tasks(filter, sort)
except:
pass
# check if there is a 'task_onhold_filter' put on the session
try:
onhold = self.request.session['task_onhold_filter']
if onhold!=None:
tasks = tasks.filter(resume=not onhold)
except:
pass
if (search_box is not None):
tasks = get_searched_tasks(search_box, sort)
paginator = Paginator(tasks, config.TASKS_PER_PAGE) # Show 50 tasks per page
page = self.request.GET.get('page')
try:
# check if there was a page on the session, if so, use it.
if page == None:
page = self.request.session['page']
self.request.session['page'] = None
except:
pass
try:
tasks = paginator.page(page)
except PageNotAnInteger:
# If page is not an integer, deliver first page.
tasks = paginator.page(1)
except EmptyPage:
# If page is out of range (e.g. 9999), deliver last page of results.
tasks = paginator.page(paginator.num_pages)
return tasks
def get_searched_tasks_only_status(search, sort):
tasks = Task.objects.filter(status__in=search).order_by(sort)
return tasks
def get_searched_tasks(search, sort):
tasks = Task.objects.filter(
Q(id__contains=search) |
Q(sas_id__contains=search) |
Q(creationTime__icontains=search) |
Q(filter__icontains=search) |
Q(status__icontains=search) |
Q(status__in=search) |
Q(project__icontains=search)).order_by(sort)
return tasks
class TaskTables2View(SingleTableView):
model = Task
table_class = TaskTable
template_name = 'query/query.html'
def TaskDetails(request, id=0, page=0):
try:
task = Task.objects.get(id=id)
# store the requested task_id on the session
request.session['task_id'] = task.id
except:
# when an invalid id is given, like '/atdb/task_details/0/',
# then look if there is a task stored on the session
try:
task_on_session = request.session['task_id']
task = Task.objects.get(id=task_on_session)
except:
messages.add_message(request, messages.WARNING, 'no task selected.')
return redirect('homepage')
# store the current page on the session
request.session['page'] = page
log_entries = LogEntry.objects.filter(task=task).order_by('-timestamp')
logentries_html = algorithms.convert_logentries_to_html(log_entries)
return render(request, "taskdatabase/tasks/task_details.html", {'task': task, 'logentries': logentries_html })
def ShowInputs(request, id):
task = Task.objects.get(id=id)
# convert the json to a presentable piece of html for the output template
results = algorithms.convert_json_to_nested_table(task.inputs)
return render(request, "taskdatabase/details/inputs.html", {'results': results})
def ShowOutputs(request, id):
task = Task.objects.get(id=id)
# convert the json to a presentable piece of html for the output template
results = algorithms.convert_json_to_nested_table(task.outputs)
return render(request, "taskdatabase/details/outputs.html", {'results': results})
def ShowMetrics(request, id):
task = Task.objects.get(id=id)
# convert the json to a presentable piece of html for the output template
results = algorithms.convert_list_of_dicts_to_html(task.metrics)
return render(request, "taskdatabase/details/metrics.html", {'results': results})
def ShowConfig(request):
configuration = Configuration.objects.all()
results = algorithms.convert_config_to_html(configuration)
return render(request, "taskdatabase/config.html", {'results': results})
def ShowDashboard(request, selection):
# gather the results
results_tasks,results_logs = algorithms.construct_dashboard_html(request, selection)
return render(request, "dashboard/dashboard.html",
{'results_tasks': results_tasks,
'results_logs': results_logs,
'selection': selection})
def WorkflowDetails(request, id):
workflow = Workflow.objects.get(id=id)
return render(request, "taskdatabase/details/workflow_details.html", {'workflow': workflow})
class DiagramView(ListView):
model = Task
template_name = "taskdatabase/diagram.html"
# ---------- REST API views -----------
# example: /atdb/tasks/
class TaskListViewAPI(generics.ListCreateAPIView):
"""
A pagination list of tasks, unsorted.
"""
model = Task
queryset = Task.objects.all().order_by('-priority','id')
#serializer_class = TaskSerializer
# using the Django Filter Backend - https://django-filter.readthedocs.io/en/latest/index.html
filter_backends = (filters.DjangoFilterBackend,)
filter_class = TaskFilter
def get_serializer_class(self):
if self.request.method in ['GET']:
return TaskReadSerializer
else:
return TaskWriteSerializer
class TaskListViewAPIFast(generics.ListAPIView):
"""
A pagination list of tasks, unsorted.
"""
model = Task
queryset = Task.objects.all().order_by('-priority','id')
serializer_class = TaskReadSerializerFast
# using the Django Filter Backend - https://django-filter.readthedocs.io/en/latest/index.html
filter_backends = (filters.DjangoFilterBackend,)
filter_class = TaskFilter
# example: /atdb/tasks/5/
# calling this view serializes a task in the REST API
class TaskDetailsViewAPI(generics.RetrieveUpdateDestroyAPIView):
"""
Detailed view of a task.
"""
model = Task
queryset = Task.objects.all()
# serializer_class = TaskSerializer
def get_serializer_class(self):
if self.request.method in ['GET']:
return TaskReadSerializer
else:
return TaskWriteSerializer
class TaskDetailsViewAPIFast(generics.RetrieveUpdateDestroyAPIView):
"""
Detailed view of a task.
"""
model = Task
queryset = Task.objects.all()
serializer_class = TaskReadSerializerFast
# example: /atdb/workflows/
class WorkflowListViewAPI(generics.ListCreateAPIView):
model = Workflow
queryset = Workflow.objects.all()
serializer_class = WorkflowSerializer
filter_backends = (filters.DjangoFilterBackend,)
filter_class = WorkflowFilter
# example: /atdb/workflows/5/
class WorkflowDetailsViewAPI(generics.RetrieveUpdateDestroyAPIView):
model = Workflow
queryset = Workflow.objects.all()
serializer_class = WorkflowSerializer
# example: /atdb/logentries/
class LogEntryListViewAPI(generics.ListCreateAPIView):
model = LogEntry
queryset = LogEntry.objects.all()
serializer_class = LogEntrySerializer
filter_backends = (filters.DjangoFilterBackend,)
filter_class = LogEntryFilter
# overriding the POST, because the status that comes in with the LogEntry
# also needs to propagate to the task.new_status
def perform_create(self, serializer):
log_entry = serializer.save()
task = log_entry.task
task.new_status = log_entry.status
task.save()
# example: /atdb/workflows/5/
# calling this view serializes a task in the REST API
class LogEntryDetailsViewAPI(generics.RetrieveUpdateDestroyAPIView):
"""
Detailed view of a LogEntry.
"""
model = LogEntry
queryset = LogEntry.objects.all()
serializer_class = LogEntrySerializer
# overriding the POST, because the status that comes in with the LogEntry
# also needs to propagate to the task.new_status
def perform_create(self, serializer):
log_entry = serializer.save()
task = log_entry.task
task.new_status = log_entry.status
task.save()
# example: /atdb/configuration/
class ConfigurationListViewAPI(generics.ListCreateAPIView):
model = Configuration
queryset = Configuration.objects.all()
serializer_class = ConfigurationSerializer
filter_backends = (filters.DjangoFilterBackend,)
filter_class = ConfigurationFilter
# example: /atdb/configuration/5/
class ConfigurationDetailsViewAPI(generics.RetrieveUpdateDestroyAPIView):
model = Configuration
queryset = Configuration.objects.all()
serializer_class = ConfigurationSerializer
# example: /atdb/job/
class JobListViewAPI(generics.ListCreateAPIView):
model = Job
queryset = Job.objects.all()
serializer_class = JobSerializer
filter_backends = (filters.DjangoFilterBackend,)
filter_class = JobFilter
# example: /atdb/job/5/
class JobDetailsViewAPI(generics.RetrieveUpdateDestroyAPIView):
model = Job
queryset = Job.objects.all()
serializer_class = JobSerializer
# --- controller resources, triggered by a button in the GUI or directoy with a URL ---
# set task status to 'new_status' - called from the GUI
@login_required
def Hold(request,pk,hold_it,page=0):
model = Task
task = Task.objects.get(pk=pk)
task.resume = (hold_it == 'resume')
task.save()
if page==0:
# redirect to details screen
return redirect('/atdb/query')
else:
# redirect to tasks list
return redirect('/atdb/?page='+page)
@login_required
def TaskSetStatus(request,pk,new_status,page=0):
model = Task
task = Task.objects.get(pk=pk)
task.new_status = new_status
task.save()
if page==0:
# redirect to details screen
return redirect('/atdb/task_details')
else:
# redirect to tasks list
return redirect('/atdb/?page='+page)
# set a filter value in the session, used later by the 'get_searched_tasks' mechanism
def TaskSetFilter(request,filter):
request.session['task_filter'] = filter
# switch off the other filters
if filter == 'all':
request.session['task_onhold_filter'] = None
return redirect('/atdb/?page=1')
# set the defined list of ACTIVE_STATUSSES on the session, used later by the 'get_searched_tasks' mechanism
def TaskSetActiveFilter(request):
request.session['task_filter'] = settings.ACTIVE_STATUSSES
request.session['task_onhold_filter'] = None
return redirect('/atdb/?page=1')
def TaskSetOnHoldFilter(request, onhold):
request.session['task_onhold_filter'] = onhold
return redirect('/atdb/?page=1')
@login_required
def ChangePriority(request, pk, priority_change, page=0):
model = Task
task = Task.objects.get(pk=pk)
priority = task.priority + int(priority_change)
if priority<0:
priority=0
task.priority = priority
task.save()
if page==0:
# redirect to details screen
return redirect('/atdb/task_details')
else:
# redirect to tasks list
return redirect('/atdb/?page='+page)
def SortTasks(request,sort):
# store the sort field on the session
request.session['sort'] = sort
return redirect('/atdb')
@login_required
def TaskSetStatusTables2(request,pk,new_status,page):
model = Task
task = Task.objects.get(pk=pk)
task.new_status = new_status
task.save()
return redirect('/atdb/query/?page='+page)
@login_required
def TaskMultiStatus(request, new_status):
# get the list of id's from the session
query_list_of_ids = request.session['query_list_of_ids']
count = len(query_list_of_ids)
if request.method == "POST":
for id in query_list_of_ids:
task = Task.objects.get(id=id[0])
task.new_status = new_status
task.save()
return redirect('/atdb/query')
return render(request, "query/confirm_multi_change.html",{'new_value': new_status, 'count' : count})
@login_required
def TaskMultiHold(request, onhold):
# get the list of id's from the session
query_list_of_ids = request.session['query_list_of_ids']
count = len(query_list_of_ids)
if request.method == "POST":
for id in query_list_of_ids:
task = Task.objects.get(id=id[0])
task.resume = (onhold == 'resume')
task.save()
return redirect('/atdb/query')
return render(request, "query/confirm_multi_change.html",{'new_value': onhold, 'count' : count})
# /atdb/get_size?status__in=defined,staged
class GetSizeView(generics.ListAPIView):
queryset = Task.objects.all()
# override list and generate a custom response
def list(self, request, *args, **kwargs):
query_params = dict(self.request.query_params)
try:
status_in = query_params['status__in']
status_list=status_in[0].split(',')
if status_list==['']:
status_list = settings.STATUSSES_WITH_DATA
except:
# if no 'status__in=' is given, then use the default list
status_list = settings.STATUSSES_WITH_DATA
try:
type = query_params['type'][0]
# should be 'processed' or 'to_process'
except:
# if no 'type=' is given, then use the default list
type = 'to_process'
size = algorithms.get_size(status_list, type)
# return a response
return Response({
'total_size': size,
})