-
Nico Vermaas authoredNico Vermaas authored
Code owners
Assign users and groups as approvers for specific file changes. Learn more.
views.py 15.40 KiB
import time
import logging
import json
import datetime
from . import config
from django.contrib.auth.decorators import login_required
from django.views.generic import ListView
from django.contrib import messages
from django.core.cache import cache
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
from .tables import TaskTable
from django.db.models import Q
from .serializers import \
TaskWriteSerializer, \
TaskReadSerializer, \
WorkflowSerializer,\
LogEntrySerializer,\
ConfigurationSerializer
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', 'lte', 'gte'],
'workflow__id': ['exact'],
'priority': ['exact', 'lte', 'gte'],
'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 = {
'key': ['exact', 'icontains'],
}
# ---------- 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()
# get list of id's
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_list_of_dicts_to_html(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_list_of_dicts_to_html(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
# 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
# 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
# --- 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):
model = Task
task = Task.objects.get(pk=pk)
task.resume = (hold_it == 'resume')
task.save()
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
return redirect('/atdb/?page=1')
def TaskSetOnHoldFilter(request, onhold):
request.session['task_onhold_filter'] = onhold
return redirect('/atdb/?page=1')
@login_required
def TaskChangePriority(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 TaskSetStatusMultiTables2(request,new_status):
# read the current querylist from the session
# yikes, this doesn't work if 2 users are simultaneously logged in
# query_list = cache.get('query_list')
# cache.delete('query_list')
# get the list of id's from the session
query_list_of_ids = request.session['query_list_of_ids']
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')
# /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
size = algorithms.get_size(status_list)
# return a response
return Response({
'total_size': size,
})