"""Views for the genotypes.""" from MySQLdb.cursors import DictCursor from gn_libs.mysqldb import database_connection from flask import (flash, request, url_for, redirect, Blueprint, render_template, current_app as app) from uploader.ui import make_template_renderer from uploader.oauth2.client import oauth2_post from uploader.authorisation import require_login from uploader.species.models import all_species, species_by_id from uploader.monadic_requests import make_either_error_handler from uploader.request_checks import with_species, with_population from uploader.datautils import safe_int, order_by_family, enumerate_sequence from uploader.population.models import (populations_by_species, population_by_species_and_id) from .models import (genotype_markers, genotype_dataset, save_new_dataset, genotype_markers_count, genocode_by_population) genotypesbp = Blueprint("genotypes", __name__) render_template = make_template_renderer("genotypes") @genotypesbp.route("populations/genotypes", methods=["GET"]) @require_login def index(): """Direct entry-point for genotypes.""" with database_connection(app.config["SQL_URI"]) as conn: if not bool(request.args.get("species_id")): return render_template("genotypes/index.html", species=order_by_family(all_species(conn)), activelink="genotypes") species = species_by_id(conn, request.args.get("species_id")) if not bool(species): flash(f"Could not find species with ID '{request.args.get('species_id')}'!", "alert-danger") return redirect(url_for("species.populations.genotypes.index")) return redirect(url_for("species.populations.genotypes.select_population", species_id=species["SpeciesId"])) @genotypesbp.route("//populations/genotypes/select-population", methods=["GET"]) @require_login @with_species(redirect_uri="species.populations.genotypes.index") def select_population(species: dict, species_id: int): """Select the population under which the genotypes go.""" with database_connection(app.config["SQL_URI"]) as conn: if not bool(request.args.get("population_id")): return render_template("genotypes/select-population.html", species=species, populations=order_by_family( populations_by_species(conn, species_id), order_key="FamilyOrder"), activelink="genotypes") population = population_by_species_and_id( conn, species_id, request.args.get("population_id")) if not bool(population): flash("Invalid population selected!", "alert-danger") return redirect(url_for( "species.populations.genotypes.select_population", species_id=species_id)) return redirect(url_for("species.populations.genotypes.list_genotypes", species_id=species_id, population_id=population["Id"])) @genotypesbp.route( "//populations//genotypes", methods=["GET"]) @require_login @with_population(species_redirect_uri="species.populations.genotypes.index", redirect_uri="species.populations.genotypes.select_population") def list_genotypes(species: dict, population: dict, **kwargs):# pylint: disable=[unused-argument] """List genotype details for species and population.""" with database_connection(app.config["SQL_URI"]) as conn: return render_template("genotypes/list-genotypes.html", species=species, population=population, genocode=genocode_by_population( conn, population["Id"]), total_markers=genotype_markers_count( conn, species["SpeciesId"]), dataset=genotype_dataset(conn, species["SpeciesId"], population["Id"]), activelink="list-genotypes") @genotypesbp.route( "//populations//genotypes/list-markers", methods=["GET"]) @require_login @with_population(species_redirect_uri="species.populations.genotypes.index", redirect_uri="species.populations.genotypes.select_population") def list_markers( species: dict, population: dict, **kwargs ):# pylint: disable=[unused-argument] """List a species' genetic markers.""" with database_connection(app.config["SQL_URI"]) as conn: start_from = max(safe_int(request.args.get("start_from") or 0), 0) count = safe_int(request.args.get("count") or 20) return render_template("genotypes/list-markers.html", species=species, population=population, total_markers=genotype_markers_count( conn, species["SpeciesId"]), start_from=start_from, count=count, markers=enumerate_sequence( genotype_markers(conn, species["SpeciesId"], offset=start_from, limit=count), start=start_from+1), activelink="list-markers") @genotypesbp.route( "//populations//genotypes/datasets/" "/view", methods=["GET"]) @require_login def view_dataset(species_id: int, population_id: int, dataset_id: int): """View details regarding a specific dataset.""" with database_connection(app.config["SQL_URI"]) as conn: species = species_by_id(conn, species_id) if not bool(species): flash("Invalid species provided!", "alert-danger") return redirect(url_for("species.populations.genotypes.index")) population = population_by_species_and_id( conn, species_id, population_id) if not bool(population): flash("Invalid population selected!", "alert-danger") return redirect(url_for( "species.populations.genotypes.select_population", species_id=species_id)) dataset = genotype_dataset(conn, species_id, population_id, dataset_id) if not bool(dataset): flash("Could not find such a dataset!", "alert-danger") return redirect(url_for( "species.populations.genotypes.list_genotypes", species_id=species_id, population_id=population_id)) return render_template("genotypes/view-dataset.html", species=species, population=population, dataset=dataset, activelink="view-dataset") @genotypesbp.route( "//populations//genotypes/datasets/" "create", methods=["GET", "POST"]) @require_login @with_population(species_redirect_uri="species.populations.genotypes.index", redirect_uri="species.populations.genotypes.select_population") def create_dataset(species: dict, population: dict, **kwargs):# pylint: disable=[unused-argument] """Create a genotype dataset.""" with (database_connection(app.config["SQL_URI"]) as conn, conn.cursor(cursorclass=DictCursor) as cursor): if request.method == "GET": return render_template("genotypes/create-dataset.html", species=species, population=population, activelink="create-dataset") form = request.form new_dataset = save_new_dataset( cursor, population["Id"], form["geno-dataset-name"], form["geno-dataset-fullname"], form["geno-dataset-shortname"]) def __success__(_success): flash("Successfully created genotype dataset.", "alert-success") return redirect(url_for( "species.populations.genotypes.list_genotypes", species_id=species["SpeciesId"], population_id=population["Id"])) return oauth2_post( "auth/resource/genotypes/create", json={ **dict(request.form), "species_id": species["SpeciesId"], "population_id": population["Id"], "dataset_id": new_dataset["Id"], "dataset_name": form["geno-dataset-name"], "dataset_fullname": form["geno-dataset-fullname"], "dataset_shortname": form["geno-dataset-shortname"], "public": "on" } ).either( make_either_error_handler( "There was an error creating the genotype dataset."), __success__)