Skip to content
Snippets Groups Projects
survey.py 8.13 KiB
import logging
from typing import Any, TypedDict, List, Dict

from flask import Blueprint
from flask_login import login_required, current_user  # type: ignore
from sqlalchemy import delete, select
from sqlalchemy.orm import joinedload, load_only

from compendium_v2.db import db
from compendium_v2.db.presentation_models import NREN, PreviewYear
from compendium_v2.db.survey_models import Survey, SurveyResponse, SurveyStatus, RESPONSE_NOT_STARTED
from compendium_v2.publishers.survey_publisher import publish
from compendium_v2.routes import common
from compendium_v2.auth.session_management import admin_required


routes = Blueprint('survey', __name__)
logger = logging.getLogger(__name__)


LIST_SURVEYS_RESPONSE_SCHEMA = {
    '$schema': 'http://json-schema.org/draft-07/schema#',
    'definitions': {
        'response': {
            'type': 'object',
            'properties': {
                'nren': {'type': 'string'},
                'status': {'type': 'string'},
                'lock_description': {'type': 'string'},
            },
            'required': ['nren', 'status', 'lock_description'],
            'additionalProperties': False
        },
        'survey': {
            'type': 'object',
            'properties': {
                'year': {'type': 'integer'},
                'status': {'type': 'string'},
                'responses': {'type': 'array', 'items': {'$ref': '#/definitions/response'}},
            },
            'required': ['year', 'status', 'responses'],
            'additionalProperties': False
        }
    },
    'type': 'array',
    'items': {'$ref': '#/definitions/survey'}
}


@routes.route('/list', methods=['GET'])
@common.require_accepts_json
@login_required
def list_surveys() -> Any:
    """
    retrieve a list of surveys and responses, including their status

    response will be formatted as:

    .. asjson::
        compendium_v2.routes.survey.LIST_SURVEYS_RESPONSE_SCHEMA

    """

    if not (current_user.is_admin or current_user.is_observer):
        return {'message': 'Insufficient privileges to access this resource'}, 403

    surveys = db.session.scalars(
        select(Survey).options(
            load_only(Survey.year, Survey.status),
            joinedload(Survey.responses).load_only(SurveyResponse.status)
            .joinedload(SurveyResponse.nren).load_only(NREN.name)
        ).order_by(Survey.year.desc())
    ).unique()

    def response_key(response):
        return response.status.value + response.nren.name.lower()

    class SurveyDict(TypedDict):
        year: int
        status: str
        responses: List[Dict[str, str]]

    entries: List[SurveyDict] = []

    def _get_response(response: SurveyResponse) -> Dict[str, str]:
        res = {
            "nren": response.nren.name,
            "status": response.status.value,
            "lock_description": response.lock_description
        }
        if current_user.is_observer:
            res["lock_description"] = response.lock_description
        return res

    for entry in surveys:
        # only include lock description if the user is an admin
        entries.append(
            {
                "year": entry.year,
                "status": entry.status.value,
                "responses": [_get_response(r) for r in sorted(entry.responses, key=response_key)]
            })

    # add in nrens without a response if the survey is open
    nren_names = set([name for name in db.session.scalars(select(NREN.name))])
    for survey_dict in entries:
        if survey_dict["status"] != SurveyStatus.open.value:
            continue
        nrens_with_responses = set([r["nren"] for r in survey_dict["responses"]])
        for nren_name in sorted(nren_names.difference(nrens_with_responses), key=str.lower):
            survey_dict["responses"].append({"nren": nren_name, "status": RESPONSE_NOT_STARTED, "lock_description": ""})

    return entries


@routes.route('/new', methods=['POST'])
@common.require_accepts_json
@admin_required
def start_new_survey() -> Any:
    """
    endpoint to initiate a new survey

    :returns: ``{'success': True}`` or a 400 or 404 status with a descriptive message
    """
    all_surveys = db.session.scalars(select(Survey).options(load_only(Survey.status)))
    if any([survey.status != SurveyStatus.published for survey in all_surveys]):
        return {'message': 'All earlier surveys should be published before starting a new one'}, 400

    last_survey = db.session.scalar(
        select(Survey).order_by(Survey.year.desc()).limit(1)
    )

    if not last_survey:
        return {'message': 'No surveys found'}, 404

    new_year = last_survey.year + 1
    new_survey = last_survey.survey
    new_survey = Survey(year=new_year, survey=new_survey, status=SurveyStatus.closed)
    db.session.add(new_survey)
    db.session.commit()

    return {'success': True}


@routes.route('/open/<int:year>', methods=['POST'])
@common.require_accepts_json
@admin_required
def open_survey(year) -> Any:
    """
    endpoint to open a survey to the nrens

    :returns: ``{'success': True}`` or a 400 or 404 status with a descriptive message
    """
    survey = db.session.scalar(select(Survey).where(Survey.year == year))
    if not survey:
        return {'message': 'Survey not found'}, 404

    if survey.status != SurveyStatus.closed:
        return {'message': 'Survey is not closed and can therefore not be opened'}, 400

    all_surveys = db.session.scalars(select(Survey))
    if any([s.status == SurveyStatus.open for s in all_surveys]):
        return {'message': 'There already is an open survey'}, 400

    survey.status = SurveyStatus.open
    db.session.commit()

    return {'success': True}


@routes.route('/close/<int:year>', methods=['POST'])
@common.require_accepts_json
@admin_required
def close_survey(year) -> Any:
    """
    endpoint to close a survey to the nrens

    :returns: ``{'success': True}`` or a 400 or 404 status with a descriptive message
    """
    survey = db.session.scalar(select(Survey).where(Survey.year == year))
    if not survey:
        return {'message': 'Survey not found'}, 404

    if survey.status != SurveyStatus.open:
        return {'message': 'Survey is not open and can therefore not be closed'}, 400

    survey.status = SurveyStatus.closed
    db.session.commit()

    return {'success': True}


@routes.route('/preview/<int:year>', methods=['POST'])
@common.require_accepts_json
@admin_required
def preview_survey(year) -> Any:
    """
    endpoint to preview a survey on the compendium website

    :returns: ``{'success': True}`` or a 400 or 404 status with a descriptive message
    """
    survey = db.session.scalar(select(Survey).where(Survey.year == year))
    if not survey:
        return {'message': 'Survey not found'}, 404

    if survey.status not in [SurveyStatus.closed, SurveyStatus.preview]:
        return {'message': 'Survey is not closed or in preview and can therefore not be published for preview'}, 400

    if year < 2023:
        return {'message': 'The 2023 survey is the first that can be published from this application'}, 400

    publish(year)

    preview = db.session.scalar(select(PreviewYear).where(PreviewYear.year == year))
    if not preview:
        db.session.add(PreviewYear(year=year))

    survey.status = SurveyStatus.preview
    db.session.commit()

    return {'success': True}


@routes.route('/publish/<int:year>', methods=['POST'])
@common.require_accepts_json
@admin_required
def publish_survey(year) -> Any:
    """
    endpoint to publish a survey to the compendium website

    :returns: ``{'success': True}`` or a 400 or 404 status with a descriptive message
    """
    survey = db.session.scalar(select(Survey).where(Survey.year == year))
    if not survey:
        return {'message': 'Survey not found'}, 404

    if survey.status not in [SurveyStatus.preview, SurveyStatus.published]:
        return {'message': 'Survey is not in preview or published and can therefore not be published'}, 400

    if year < 2023:
        return {'message': 'The 2023 survey is the first that can be published from this application'}, 400

    publish(year)

    db.session.execute(delete(PreviewYear).where(PreviewYear.year == year))

    survey.status = SurveyStatus.published
    db.session.commit()

    return {'success': True}