Select Git revision
survey.py 9.00 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'}
}
SURVEY_ACTIVE_YEAR_RESPONSE_SCHEMA = {
"$schema": "http://json-schema.org/draft-07/schema#",
"type": "object",
"properties": {
"year": {
"type": "string"
}
},
"required": ["year"],
"additionalProperties": False
}
@routes.route('/active/year', methods=['GET'])
@common.require_accepts_json
@login_required
def get_active_survey_year() -> Any:
"""
retrieve a year of latest active survey
response will be formatted as:
.. asjson::
compendium_v2.routes.survey.SURVEY_ACTIVE_YEAR_RESPONSE_SCHEMA
"""
survey_record = db.session.query(Survey.year).filter(
Survey.status == SurveyStatus.open
).order_by(Survey.year.desc()).first()
if survey_record:
year = survey_record.year
return {'year': year}
else:
return {'message': 'No open survey found.'}, 404
@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}