83 lines
3.1 KiB
Python
83 lines
3.1 KiB
Python
from datetime import datetime
|
|
from flask import request, render_template, session
|
|
from sqlalchemy import desc, asc
|
|
|
|
from common.models.document import DocumentVersion, Document
|
|
from common.utils.filtered_list_view import FilteredListView
|
|
from common.utils.view_assistants import prepare_table_for_macro
|
|
|
|
|
|
class DocumentVersionListView(FilteredListView):
|
|
allowed_filters = ['file_type', 'processing', 'processing_error']
|
|
allowed_sorts = ['id', 'processing_started_at', 'processing_finished_at', 'processing_error']
|
|
|
|
def get_query(self):
|
|
return DocumentVersion.query.join(Document)
|
|
|
|
def apply_filters(self, query):
|
|
filters = request.args.to_dict()
|
|
|
|
if filters.get('file_type'):
|
|
query = query.filter(DocumentVersion.file_type == filters['file_type'])
|
|
|
|
if filters.get('processing'):
|
|
query = query.filter(DocumentVersion.processing == (filters['processing'] == 'true'))
|
|
|
|
if filters.get('processing_error'):
|
|
if filters['processing_error'] == 'true':
|
|
query = query.filter(DocumentVersion.processing_error.isnot(None))
|
|
elif filters['processing_error'] == 'false':
|
|
query = query.filter(DocumentVersion.processing_error.is_(None))
|
|
|
|
if filters.get('start_date'):
|
|
query = query.filter(
|
|
DocumentVersion.processing_started_at >= datetime.strptime(filters['start_date'], '%Y-%m-%d'))
|
|
|
|
if filters.get('end_date'):
|
|
query = query.filter(
|
|
DocumentVersion.processing_finished_at <= datetime.strptime(filters['end_date'], '%Y-%m-%d'))
|
|
|
|
return query
|
|
|
|
def apply_sorting(self, query):
|
|
sort_by = request.args.get('sort_by', 'id')
|
|
sort_order = request.args.get('sort_order', 'asc')
|
|
|
|
if sort_by in self.allowed_sorts:
|
|
column = getattr(DocumentVersion, sort_by)
|
|
if sort_order == 'asc':
|
|
query = query.order_by(asc(column))
|
|
elif sort_order == 'desc':
|
|
query = query.order_by(desc(column))
|
|
|
|
return query
|
|
|
|
def get(self):
|
|
query = self.get_query()
|
|
query = self.apply_filters(query)
|
|
query = self.apply_sorting(query)
|
|
pagination = self.paginate(query)
|
|
|
|
rows = prepare_table_for_macro(
|
|
pagination.items,
|
|
[('id', ''), ('file_type', ''), ('processing', ''),
|
|
('processing_started_at', ''), ('processing_finished_at', ''),
|
|
('processing_error', '')]
|
|
)
|
|
|
|
context = {
|
|
'rows': rows,
|
|
'pagination': pagination,
|
|
'filters': request.args.to_dict(),
|
|
'sort_by': request.args.get('sort_by', 'id'),
|
|
'sort_order': request.args.get('sort_order', 'asc'),
|
|
'filter_options': self.get_filter_options()
|
|
}
|
|
return render_template(self.template, **context)
|
|
|
|
def get_filter_options(self):
|
|
return {
|
|
'file_type': [('pdf', 'PDF'), ('docx', 'DOCX')],
|
|
'processing': [('true', 'Processing'), ('false', 'Not Processing')],
|
|
'processing_error': [('true', 'With Errors'), ('false', 'Without Errors')]
|
|
} |