"""Endpoints handling species.""" from pymonad.either import Left, Right, Either from gn_libs.mysqldb import database_connection from flask import (flash, request, url_for, redirect, Blueprint, current_app as app) from uploader.population import popbp from uploader.platforms import platformsbp from uploader.ui import make_template_renderer from uploader.oauth2.client import oauth2_get, oauth2_post from uploader.authorisation import require_login, require_token from uploader.datautils import order_by_family, enumerate_sequence from .models import (all_species, save_species, species_by_id, update_species, species_families) speciesbp = Blueprint("species", __name__) speciesbp.register_blueprint(popbp, url_prefix="/") speciesbp.register_blueprint(platformsbp, url_prefix="/") render_template = make_template_renderer("species") @speciesbp.route("/", methods=["GET"]) @require_login def list_species(): """List and display all the species in the database.""" with database_connection(app.config["SQL_URI"]) as conn: return render_template("species/list-species.html", allspecies=enumerate_sequence(all_species(conn))) @speciesbp.route("/", methods=["GET"]) @require_login def view_species(species_id: int): """View details of a particular species and menus to act upon it.""" with database_connection(app.config["SQL_URI"]) as conn: species = species_by_id(conn, species_id) if bool(species): return render_template("species/view-species.html", species=species, activelink="view-species") flash("Could not find a species with the given identifier.", "alert-danger") return redirect(url_for("species.view_species")) @speciesbp.route("/create", methods=["GET", "POST"]) @require_login def create_species(): """Create a new species.""" # We can use uniprot's API to fetch the details with something like # https://rest.uniprot.org/taxonomy/ e.g. # https://rest.uniprot.org/taxonomy/6239 with (database_connection(app.config["SQL_URI"]) as conn, conn.cursor() as cursor): if request.method == "GET": return render_template("species/create-species.html", families=species_families(conn), activelink="create-species") error = False taxon_id = request.form.get("species_taxonomy_id", "").strip() or None common_name = request.form.get("common_name", "").strip() if not bool(common_name): flash("The common species name MUST be provided.", "alert-danger") error = True scientific_name = request.form.get("scientific_name", "").strip() if not bool(scientific_name): flash("The species' scientific name MUST be provided.", "alert-danger") error = True parts = tuple(name.strip() for name in scientific_name.split(" ")) if len(parts) != 2 or not all(bool(name) for name in parts): flash("The scientific name you provided is invalid.", "alert-danger") error = True cursor.execute( "SELECT * FROM Species WHERE FullName=%s", (scientific_name,)) res = cursor.fetchone() if bool(res): flash("A species already exists with the provided scientific name.", "alert-danger") error = True family = request.form.get("species_family", "").strip() if not bool(family): flash("The species' family MUST be selected.", "alert-danger") error = True if bool(taxon_id): cursor.execute( "SELECT * FROM Species WHERE TaxonomyId=%s", (taxon_id,)) res = cursor.fetchone() if bool(res): flash("A species already exists with the provided scientific name.", "alert-danger") error = True if error: return redirect(url_for("species.create_species", common_name=common_name, scientific_name=scientific_name, taxon_id=taxon_id)) species = save_species( conn, common_name, scientific_name, family, taxon_id) flash("Species saved successfully!", "alert-success") return redirect(url_for("species.view_species", species_id=species["species_id"])) @speciesbp.route("//edit-extra", methods=["GET", "POST"]) @require_login @require_token #def edit_species(species_id: int): def edit_species_extra(token: dict, species_id: int):# pylint: disable=[unused-argument] """Edit a species' details. Parameters ---------- token: A JWT token used for authorisation. species_id: An identifier for the species being edited. """ def __failure__(res): app.logger.debug( "There was an error in the attempt to edit the species: %s", res) flash(res, "alert-danger") return redirect(url_for("species.view_species", species_id=species_id)) def __system_resource_uuid__(resources) -> Either: sys_res = [ resource for resource in resources if resource["resource_category"]["resource_category_key"] == "system" ] if len(sys_res) != 1: return Left("Could not find/identify a valid system resource.") return Right(sys_res[0]["resource_id"]) def __check_privileges__(authorisations): if len(authorisations.items()) != 1: return Left("Got authorisations for more than a single resource!") auths = tuple(authorisations.items())[0][1] authorised = "system:species:edit-extra-info" in tuple( privilege["privilege_id"] for role in auths["roles"] for privilege in role["privileges"]) if authorised: return Right(authorised) return Left("You are not authorised to edit species extra details.") with database_connection(app.config["SQL_URI"]) as conn: species = species_by_id(conn, species_id) all_the_species = all_species(conn) families = species_families(conn) family_order = tuple( item[0] for item in order_by_family(all_the_species) if item[0][1] is not None) if bool(species) and request.method == "GET": return oauth2_get("auth/user/resources").then( __system_resource_uuid__ ).then( lambda resource_id: oauth2_post( "auth/resource/authorisation", json={"resource-ids": [resource_id]}) ).then(__check_privileges__).then( lambda authorisations: render_template( "species/edit-species.html", species=species, families=families, family_order=family_order, max_order_id = max( row["OrderId"] for row in all_the_species if row["OrderId"] is not None), activelink="edit-species") ).either(__failure__, lambda res: res) if bool(species) and request.method == "POST": update_species(conn, species_id, request.form["species_name"], request.form["species_fullname"], request.form["species_family"], int(request.form["species_familyorderid"]), int(request.form["species_orderid"])) flash("Updated species successfully.", "alert-success") return redirect(url_for("species.edit_species_extra", species_id=species_id)) flash("Species with the given identifier was not found!", "alert-danger") return redirect(url_for("species.list_species"))