def annotate()

in data_annotation_platform/app/main/routes.py [0:0]

• 75 lines of code
• 16 McCabe index (conditional complexity)

def annotate(task_id):
    if request.method == "POST":
        # record post time
        now = datetime.datetime.utcnow()

        # get the json from the client
        annotation = request.get_json()
        if annotation["identifier"] != task_id:
            flash("Internal error: task id doesn't match.", "error")
            return redirect(url_for(task_id=task_id))

        task = Task.query.filter_by(id=task_id).first()

        # if it is a re-annotation, get back to the home page
        # otherwise proceed with the next time series
        if task.done:
            next = "main.index"
        else:
            next = "main.assign"

        # remove all previous annotations for this task
        for ann in Annotation.query.filter_by(task_id=task_id).all():
            db.session.delete(ann)
        task.done = False
        task.annotated_on = None
        db.session.commit()

        # record the annotation
        if annotation["changepoints"] is None:
            ann = Annotation(cp_index=None, type=None, task_id=task_id)
            db.session.add(ann)
            db.session.commit()
        else:
            for cp in annotation["changepoints"]:
                ann = Annotation(cp_index=cp["x"], type=cp["t"], task_id=task_id)
                db.session.add(ann)
                db.session.commit()

        # mark the task as done
        task.done = True
        task.annotated_on = now
        task.time_spent = annotation["time_spent"]
        task.difficulty = annotation["difficulty"]
        task.problem = annotation["problem"]
        db.session.commit()
        done, _, todo = __get_done_and_todo(current_user)
        flash("Your annotation has been recorded, thank you! Done {}, {} to go!"\
                .format(len(done), len(todo)), "success")

        # send the annotation as email to the admin for backup
        record = {
            "user_id": task.annotator_id,
            "dataset_name": task.dataset.name,
            "dataset_id": task.dataset_id,
            "task_id": task.id,
            "annotations_raw": annotation,
        }
        send_annotation_backup(record)

        return url_for(next)

    task = Task.query.filter_by(id=task_id).first()

    # check if task exists
    if task is None:
        flash("No task with id %r exists." % task_id, "error")
        return redirect(url_for("main.index"))

    # check if task is assigned to this user
    if not task.annotator_id == current_user.id:
        flash(
            "No task with id %r has been assigned to you." % task_id, "error"
        )
        return redirect(url_for("main.index"))

    # check if task is not already done
    '''
    if task.done:
        flash("It's not possible to edit annotations at the moment.")
        return redirect(url_for("main.index"))
    '''

    data = load_data_for_chart(task.dataset.name, task.dataset.md5sum)
    if data is None:
        flash(
            "An internal error occurred loading this dataset, the admin has been notified. Please try again later. We apologise for the inconvenience.",
            "error",
        )
        return redirect(url_for("main.index"))
    title = f"Dataset: {task.dataset.id}"
    is_multi = len(data["chart_data"]["values"]) > 1
    return render_template(
        "annotate/index.html",
        title=title,
        identifier=task.id,
        data=data,
        rubric=RUBRIC,
        is_multi=is_multi,
    )