"""Views handling ('classical') phenotypes.""" from functools import wraps from requests.models import Response from MySQLdb.cursors import DictCursor from flask import (flash, request, url_for, redirect, Blueprint, render_template, current_app as app) from uploader.oauth2.client import oauth2_post from uploader.authorisation import require_login from uploader.db_utils import database_connection 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 uploader.input_validation import (encode_errors, decode_errors, is_valid_representative_name) from .models import (dataset_by_id, phenotype_by_id, phenotypes_count, save_new_dataset, dataset_phenotypes, datasets_by_population) phenotypesbp = Blueprint("phenotypes", __name__) @phenotypesbp.route("/phenotypes", methods=["GET"]) @require_login def index(): """Direct entry-point for phenotypes data handling.""" with database_connection(app.config["SQL_URI"]) as conn: if not bool(request.args.get("species_id")): return render_template("phenotypes/index.html", species=order_by_family(all_species(conn)), activelink="phenotypes") species = species_by_id(conn, request.args.get("species_id")) if not bool(species): flash("No such species!", "alert-danger") return redirect(url_for("species.populations.phenotypes.index")) return redirect(url_for("species.populations.phenotypes.select_population", species_id=species["SpeciesId"])) @phenotypesbp.route("/phenotypes/select-population", methods=["GET"]) @require_login @with_species(redirect_uri="species.populations.phenotypes.index") def select_population(species: dict, **kwargs):# pylint: disable=[unused-argument] """Select the population for your phenotypes.""" with database_connection(app.config["SQL_URI"]) as conn: if not bool(request.args.get("population_id")): return render_template("phenotypes/select-population.html", species=species, populations=order_by_family( populations_by_species( conn, species["SpeciesId"]), order_key="FamilyOrder"), activelink="phenotypes") population = population_by_species_and_id( conn, species["SpeciesId"], int(request.args["population_id"])) if not bool(population): flash("No such population found!", "alert-danger") return redirect(url_for( "species.populations.phenotypes.select_population", species_id=species["SpeciesId"])) return redirect(url_for("species.populations.phenotypes.list_datasets", species_id=species["SpeciesId"], population_id=population["Id"])) @phenotypesbp.route( "/populations//phenotypes/datasets", methods=["GET"]) @require_login @with_population(species_redirect_uri="species.populations.phenotypes.index", redirect_uri="species.populations.phenotypes.select_population") def list_datasets(species: dict, population: dict, **kwargs):# pylint: disable=[unused-argument] """List available phenotype datasets.""" with database_connection(app.config["SQL_URI"]) as conn: return render_template("phenotypes/list-datasets.html", species=species, population=population, datasets=datasets_by_population( conn, species["SpeciesId"], population["Id"]), activelink="list-datasets") def with_dataset( species_redirect_uri: str, population_redirect_uri: str, redirect_uri: str ): """Ensure the dataset actually exists.""" def __decorator__(func): @wraps(func) @with_population(species_redirect_uri, population_redirect_uri) def __with_dataset__(**kwargs): try: _spcid = int(kwargs["species_id"]) _popid = int(kwargs["population_id"]) _dsetid = int(kwargs.get("dataset_id")) select_dataset_uri = redirect(url_for( redirect_uri, species_id=_spcid, population_id=_popid)) if not bool(_dsetid): flash("You need to select a valid 'dataset_id' value.", "alert-danger") return select_dataset_uri with database_connection(app.config["SQL_URI"]) as conn: dataset = dataset_by_id(conn, _spcid, _popid, _dsetid) if not bool(dataset): flash("You must select a valid dataset.", "alert-danger") return select_dataset_uri except ValueError as _verr: app.logger.debug( "Exception converting 'dataset_id' to integer: %s", kwargs.get("dataset_id"), exc_info=True) flash("Expected 'dataset_id' value to be an integer." "alert-danger") return select_dataset_uri return func(dataset=dataset, **kwargs) return __with_dataset__ return __decorator__ @phenotypesbp.route( "/populations//phenotypes/datasets" "//view", methods=["GET"]) @require_login @with_dataset( species_redirect_uri="species.populations.phenotypes.index", population_redirect_uri="species.populations.phenotypes.select_population", redirect_uri="species.populations.phenotypes.list_datasets") def view_dataset(# pylint: disable=[unused-argument] species: dict, population: dict, dataset: dict, **kwargs): """View a specific dataset""" with database_connection(app.config["SQL_URI"]) as conn: dataset = dataset_by_id( conn, species["SpeciesId"], population["Id"], dataset["Id"]) if not bool(dataset): flash("Could not find such a phenotype dataset!", "alert-danger") return redirect(url_for( "species.populations.phenotypes.list_datasets", species_id=species["SpeciesId"], population_id=population["Id"])) start_at = max(safe_int(request.args.get("start_at") or 0), 0) count = int(request.args.get("count") or 20) return render_template("phenotypes/view-dataset.html", species=species, population=population, dataset=dataset, phenotype_count=phenotypes_count( conn, population["Id"], dataset["Id"]), phenotypes=enumerate_sequence( dataset_phenotypes(conn, population["Id"], dataset["Id"], offset=start_at, limit=count), start=start_at+1), start_from=start_at, count=count, activelink="view-dataset") @phenotypesbp.route( "/populations//phenotypes/datasets" "//phenotype/", methods=["GET"]) @require_login @with_dataset( species_redirect_uri="species.populations.phenotypes.index", population_redirect_uri="species.populations.phenotypes.select_population", redirect_uri="species.populations.phenotypes.list_datasets") def view_phenotype(# pylint: disable=[unused-argument] species: dict, population: dict, dataset: dict, xref_id: int, **kwargs ): """View an individual phenotype from the dataset.""" def __render__(privileges): return render_template( "phenotypes/view-phenotype.html", species=species, population=population, dataset=dataset, phenotype=phenotype_by_id(conn, species["SpeciesId"], population["Id"], dataset["Id"], xref_id), privileges=(privileges ### For demo! Do not commit this part + ("group:resource:edit-resource", "group:resource:delete-resource",) ### END: For demo! Do not commit this part ), activelink="view-phenotype") def __fail__(error): if isinstance(error, Response) and error.json() == "No linked resource!": return __render__(tuple()) return make_either_error_handler( "There was an error fetching the roles and privileges.")(error) with database_connection(app.config["SQL_URI"]) as conn: return oauth2_post( "/auth/resource/phenotypes/individual/linked-resource", json={ "species_id": species["SpeciesId"], "population_id": population["Id"], "dataset_id": dataset["Id"], "xref_id": xref_id } ).then( lambda resource: tuple( privilege["privilege_id"] for role in resource["roles"] for privilege in role["privileges"]) ).then(__render__).either(__fail__, lambda resp: resp) @phenotypesbp.route( "/populations//phenotypes/datasets/create", methods=["GET", "POST"]) @require_login @with_population( species_redirect_uri="species.populations.phenotypes.index", redirect_uri="species.populations.phenotypes.select_population") def create_dataset(species: dict, population: dict, **kwargs):# pylint: disable=[unused-argument] """Create a new phenotype dataset.""" with (database_connection(app.config["SQL_URI"]) as conn, conn.cursor(cursorclass=DictCursor) as cursor): if request.method == "GET": return render_template("phenotypes/create-dataset.html", activelink="create-dataset", species=species, population=population, **decode_errors( request.args.get("error_values", ""))) form = request.form _errors: tuple[tuple[str, str], ...] = tuple() if not is_valid_representative_name( (form.get("dataset-name") or "").strip()): _errors = _errors + (("dataset-name", "Invalid dataset name."),) if not bool((form.get("dataset-fullname") or "").strip()): _errors = _errors + (("dataset-fullname", "You must provide a value for 'Full Name'."),) if bool(_errors) > 0: return redirect(url_for( "species.populations.phenotypes.create_dataset", species_id=species["SpeciesId"], population_id=population["Id"], error_values=encode_errors(_errors, form))) dataset_shortname = ( form["dataset-shortname"] or form["dataset-name"]).strip() _pheno_dataset = save_new_dataset( cursor, population["Id"], form["dataset-name"].strip(), form["dataset-fullname"].strip(), dataset_shortname) return redirect(url_for("species.populations.phenotypes.list_datasets", species_id=species["SpeciesId"], population_id=population["Id"])) @phenotypesbp.route( "/populations//phenotypes/datasets" "//add-phenotypes", methods=["GET", "POST"]) @require_login @with_dataset( species_redirect_uri="species.populations.phenotypes.index", population_redirect_uri="species.populations.phenotypes.select_population", redirect_uri="species.populations.phenotypes.list_datasets") def add_phenotypes(species: dict, population: dict, dataset: dict, **kwargs):# pylint: disable=[unused-argument, too-many-locals] """Add one or more phenotypes to the dataset.""" with (database_connection(app.config["SQL_URI"]) as conn, conn.cursor(cursorclass=DictCursor) as cursor): if request.method == "GET": return render_template("phenotypes/add-phenotypes.html", species=species, population=population, dataset=dataset, activelink="add-phenotypes") raise NotImplementedError("Please implement this...")