diff options
Diffstat (limited to 'uploader/population/views.py')
| -rw-r--r-- | uploader/population/views.py | 222 |
1 files changed, 222 insertions, 0 deletions
diff --git a/uploader/population/views.py b/uploader/population/views.py new file mode 100644 index 0000000..3638453 --- /dev/null +++ b/uploader/population/views.py @@ -0,0 +1,222 @@ +"""Views dealing with populations/inbredsets""" +import re +import json +import base64 + +from MySQLdb.cursors import DictCursor +from flask import (flash, + request, + url_for, + redirect, + Blueprint, + current_app as app) + +from uploader.samples.views import samplesbp +from uploader.oauth2.client import oauth2_post +from uploader.ui import make_template_renderer +from uploader.authorisation import require_login +from uploader.genotypes.views import genotypesbp +from uploader.db_utils import database_connection +from uploader.datautils import enumerate_sequence +from uploader.phenotypes.views import phenotypesbp +from uploader.expression_data.views import exprdatabp +from uploader.monadic_requests import make_either_error_handler +from uploader.species.models import (all_species, + species_by_id, + order_species_by_family) + +from .models import (save_population, + population_families, + populations_by_species, + population_genetic_types, + population_by_species_and_id) + +__active_link__ = "populations" +popbp = Blueprint("populations", __name__) +popbp.register_blueprint(samplesbp, url_prefix="/") +popbp.register_blueprint(genotypesbp, url_prefix="/") +popbp.register_blueprint(phenotypesbp, url_prefix="/") +popbp.register_blueprint(exprdatabp, url_prefix="/") +render_template = make_template_renderer("populations") + + +@popbp.route("/populations", methods=["GET", "POST"]) +@require_login +def index(): + """Entry point for populations.""" + with database_connection(app.config["SQL_URI"]) as conn: + if not bool(request.args.get("species_id")): + return render_template( + "populations/index.html", + species=order_species_by_family(all_species(conn))) + species = species_by_id(conn, request.args.get("species_id")) + if not bool(species): + flash("Invalid species identifier provided!", "alert-danger") + return redirect(url_for("species.populations.index")) + return redirect(url_for("species.populations.list_species_populations", + species_id=species["SpeciesId"])) + +@popbp.route("/<int:species_id>/populations", methods=["GET"]) +@require_login +def list_species_populations(species_id: int): + """List a particular species' populations.""" + with database_connection(app.config["SQL_URI"]) as conn: + species = species_by_id(conn, species_id) + if not bool(species): + flash("No species was found for given ID.", "alert-danger") + return redirect(url_for("species.populations.index")) + return render_template( + "populations/list-populations.html", + species=species, + populations=enumerate_sequence(populations_by_species( + conn, species_id)), + activelink="list-populations") + + +def valid_population_name(population_name: str) -> bool: + """ + Check whether the given name is a valid population name. + + Parameters + ---------- + population_name: a string of characters. + + Checks For + ---------- + * The name MUST start with an alphabet [a-zA-Z] + * The name MUST end with an alphabet [a-zA-Z] or number [0-9] + * The name MUST be composed of alphabets [a-zA-Z], numbers [0-9], + underscores (_) and/or hyphens (-). + + Returns + ------- + Boolean indicating whether or not the name is valid. + """ + pattern = re.compile(r"^[a-zA-Z]+[a-zA-Z0-9_-]*[a-zA-Z0-9]$") + return bool(pattern.match(population_name)) + + +@popbp.route("/<int:species_id>/populations/create", methods=["GET", "POST"]) +@require_login +def create_population(species_id: int): + """Create a new population.""" + with (database_connection(app.config["SQL_URI"]) as conn, + conn.cursor(cursorclass=DictCursor) as cursor): + species = species_by_id(conn, species_id) + + if request.method == "GET": + error_values = request.args.get("error_values") + if not bool(error_values): + error_values = base64.b64encode( + '{"errors":{}, "error_values": {}}'.encode("utf8") + ).decode("utf8") + + error_values = json.loads(base64.b64decode( + error_values.encode("utf8")).decode("utf8"))# type: ignore[union-attr] + return render_template( + "populations/create-population.html", + species=species, + families = population_families(conn), + genetic_types = population_genetic_types(conn), + mapping_methods=( + {"id": "0", "value": "No mapping support"}, + {"id": "1", "value": "GEMMA, QTLReaper, R/qtl"}, + {"id": "2", "value": "GEMMA"}, + {"id": "3", "value": "R/qtl"}, + {"id": "4", "value": "GEMMA, PLINK"}), + activelink="create-population", + **error_values) + + if not bool(species): + flash("You must select a species.", "alert-danger") + return redirect(url_for("species.populations.index")) + + errors: tuple[tuple[str, str], ...] = tuple() + + population_name = (request.form.get( + "population_name") or "").strip() + if not bool(population_name): + errors = errors + (("population_name", + "You must provide a name for the population!"),) + + if not valid_population_name(population_name): + errors = errors + (( + "population_name", + "The population name can only contain letters, numbers, " + "hyphens and underscores."),) + + population_fullname = (request.form.get( + "population_fullname") or "").strip() + if not bool(population_fullname): + errors = errors + ( + ("population_fullname", "Full Name MUST be provided."),) + + if bool(errors): + values = base64.b64encode( + json.dumps({ + "errors": dict(errors), + "error_values": dict(request.form) + }).encode("utf8")) + return redirect(url_for("species.populations.create_population", + species_id=species["SpeciesId"], + error_values=values)) + + new_population = save_population(cursor, { + "SpeciesId": species["SpeciesId"], + "Name": population_name, + "InbredSetName": population_fullname, + "FullName": population_fullname, + "InbredSetCode": request.form.get("population_code") or None, + "Description": request.form.get("population_description") or None, + "Family": request.form.get("population_family") or None, + "MappingMethodId": request.form.get("population_mapping_method_id"), + "GeneticType": request.form.get("population_genetic_type") or None + }) + + def __flash_success__(_success): + flash("Successfully created resource.", "alert-success") + return redirect(url_for( + "species.populations.view_population", + species_id=species["SpeciesId"], + population_id=new_population["InbredSetId"])) + + app.logger.debug("We begin setting up the privileges here…") + return oauth2_post( + "auth/resource/populations/create", + json={ + **dict(request.form), + "species_id": species_id, + "population_id": new_population["Id"], + "public": "on" + } + ).either( + make_either_error_handler( + "There was an error creating the population"), + __flash_success__) + + +@popbp.route("/<int:species_id>/populations/<int:population_id>", + methods=["GET"]) +@require_login +def view_population(species_id: int, population_id: int): + """View the details of a population.""" + with database_connection(app.config["SQL_URI"]) as conn: + species = species_by_id(conn, species_id) + population = population_by_species_and_id(conn, species_id, population_id) + error = False + + if not bool(species): + flash("You must select a species.", "alert-danger") + error = True + + if not bool(population): + flash("You must select a population.", "alert-danger") + error = True + + if error: + return redirect(url_for("species.populations.index")) + + return render_template("populations/view-population.html", + species=species, + population=population, + activelink="view-population") |