Source code for dokomoforms.handlers.api.v0.submissions

"""TornadoResource class for dokomoforms.models.submission.Submission."""
from contextlib import closing
from csv import DictWriter
from io import StringIO
from itertools import chain

import restless.exceptions as exc

from dokomoforms.handlers.api.v0 import BaseResource
from dokomoforms.models import (
    Survey, Submission, User,
    construct_submission, construct_answer, Answer,
    SurveyNode, skipped_required,
    get_model
)
from dokomoforms.models.answer import ANSWER_TYPES
from dokomoforms.exc import RequiredQuestionSkipped


def _create_answer(session, answer_dict) -> Answer:
    survey_node_id = answer_dict['survey_node_id']
    error = exc.BadRequest('survey_node not found: {}'.format(survey_node_id))
    survey_node = get_model(session, SurveyNode, survey_node_id, error)
    answer_dict['survey_node'] = survey_node
    return construct_answer(**answer_dict)


def _create_submission(self, survey):
    # Unauthenticated submissions are only allowed if the survey_type is
    # 'public'.
    authenticated = super(self.__class__, self).is_authenticated()
    if not authenticated:
        if survey.survey_type == 'public':
            self._check_xsrf_cookie()
        else:
            raise exc.Unauthorized()

    # If logged in, add enumerator
    if self.current_user_model is not None:
        try:
            enumerator = self._get_model(
                self.data['enumerator_user_id'], model_cls=User
            )
        except KeyError:
            self.data['enumerator'] = self.current_user_model
        else:
            self.data['enumerator'] = enumerator

    self.data['survey'] = survey

    with self.session.begin():
        # create a list of Answer models
        if 'answers' in self.data:
            raw_answers = self.data['answers']
            answers = [
                _create_answer(self.session, answer) for answer in raw_answers
            ]
            self.data['answers'] = answers

        self.data['submission_type'] = survey.survey_type + '_submission'

        submission = construct_submission(**self.data)

        # add the answer models
        # if 'answers' in self.data:
        #    submission.answers = answers

        # add the submission
        self.session.add(submission)
        self.session.flush()

        for answer in submission.answers:
            answer.main_answer = (
                self.session
                .query(ANSWER_TYPES[answer.answer_type].main_answer)
                .filter_by(id=answer.id)
                .scalar()
            )

        skipped_question = skipped_required(survey, submission.answers)
        if skipped_question is not None:
            raise RequiredQuestionSkipped(
                '{} skipped'.format(skipped_question)
            )

    return submission


[docs]class SubmissionResource(BaseResource): """Restless resource for Submissions. BaseResource sets the serializer, which uses the dokomo models' ModelJSONEncoder for json conversion. """ # Set the property name on the outputted json resource_type = Submission default_sort_column_name = 'save_time' objects_key = 'submissions' def _csv(self, raw_answers) -> dict: """Return {'format': 'csv', 'data': <csv-formatted string>}.""" answers = [answer._asdict('csv') for answer in raw_answers] dialect = self._query_arg('dialect', default='excel') fieldnames = [ 'id', 'deleted', 'answer_number', 'submission_id', 'save_time', 'survey_id', 'survey_node_id', 'question_id', 'type_constraint', 'last_update_time', 'main_answer', 'response', 'response_type', 'metadata' ] with closing(StringIO()) as out: dw = DictWriter(out, fieldnames=fieldnames, dialect=dialect) dw.writeheader() dw.writerows(answers) return {'format': 'csv', 'data': out.getvalue()}
[docs] def wrap_list_response(self, data): """Allow CSV export of submission data. This method adds CSV export functionality on top of the JSON list wrapping of BaseResource.wrap_list_response. """ if self.content_type == 'csv': self._set_filename('submissions', 'csv') sub_data = data[2] raw_answers = chain.from_iterable(sub.answers for sub in sub_data) return self._csv(raw_answers) return super().wrap_list_response(data)
[docs] def is_authenticated(self): """Allow unauthenticated POSTs under the right circumstances.""" if self.request_method() == 'POST': # At this point in the lifecycle of the request, self.data has # not been populated, so we need to handle POST authentication # in the create method. return True return super().is_authenticated()
[docs] def detail(self, submission_id): """Allow CSV export of a single submission.""" if self.content_type == 'csv': self._set_filename('submission_{}'.format(submission_id), 'csv') return self._csv(self._get_model(submission_id).answers) return super().detail(submission_id)
# POST /api/submissions/
[docs] def create(self): """Create a new submission. Uses the current_user_model (i.e. logged-in user) as creator. """ survey_id = self.data.pop('survey_id') error = exc.BadRequest( 'The survey could not be found: {}'.format(survey_id) ) survey = self._get_model(survey_id, model_cls=Survey, exception=error) return _create_submission(self, survey)
[docs]def get_submission_for_handler(tornado_handler, submission_id): """Maybe a handler needs a submission from the API.""" submission_resource = SubmissionResource() submission_resource.ref_rh = tornado_handler submission_resource.request = tornado_handler.request submission_resource.application = tornado_handler.application return submission_resource.detail(submission_id)