"""Views dealing with populations/inbredsets""" import json import base64 from MySQLdb.cursors import DictCursor from gn_libs.mysqldb import database_connection 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.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.input_validation import is_valid_representative_name 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("//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") @popbp.route("//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 is_valid_representative_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("//populations/", 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")