diff options
Diffstat (limited to 'uploader/platforms')
| -rw-r--r-- | uploader/platforms/__init__.py | 2 | ||||
| -rw-r--r-- | uploader/platforms/models.py | 96 | ||||
| -rw-r--r-- | uploader/platforms/views.py | 118 |
3 files changed, 216 insertions, 0 deletions
diff --git a/uploader/platforms/__init__.py b/uploader/platforms/__init__.py new file mode 100644 index 0000000..8cb89c9 --- /dev/null +++ b/uploader/platforms/__init__.py @@ -0,0 +1,2 @@ +"""Module to handle management of genetic platforms.""" +from .views import platformsbp diff --git a/uploader/platforms/models.py b/uploader/platforms/models.py new file mode 100644 index 0000000..0dd9368 --- /dev/null +++ b/uploader/platforms/models.py @@ -0,0 +1,96 @@ +"""Handle db interactions for platforms.""" +from typing import Optional + +import MySQLdb as mdb +from MySQLdb.cursors import Cursor, DictCursor + +def platforms_by_species( + conn: mdb.Connection, + speciesid: int, + offset: int = 0, + limit: Optional[int] = None +) -> tuple[dict, ...]: + """Retrieve platforms by the species""" + _query = ("SELECT * FROM GeneChip WHERE SpeciesId=%s " + "ORDER BY GeneChipName ASC") + if bool(limit) and limit > 0:# type: ignore[operator] + _query = f"{_query} LIMIT {limit} OFFSET {offset}" + with conn.cursor(cursorclass=DictCursor) as cursor: + cursor.execute(_query, (speciesid,)) + return tuple(dict(row) for row in cursor.fetchall()) + + +def species_platforms_count(conn: mdb.Connection, species_id: int) -> int: + """Get the number of platforms in the database for a particular species.""" + with conn.cursor(cursorclass=DictCursor) as cursor: + cursor.execute( + "SELECT COUNT(GeneChipName) AS count FROM GeneChip " + "WHERE SpeciesId=%s", + (species_id,)) + return int(cursor.fetchone()["count"]) + + +def platform_by_id(conn: mdb.Connection, platformid: int) -> Optional[dict]: + """Retrieve a platform by its ID""" + with conn.cursor(cursorclass=DictCursor) as cursor: + cursor.execute("SELECT * FROM GeneChip WHERE Id=%s", + (platformid,)) + result = cursor.fetchone() + if bool(result): + return dict(result) + + return None + + +def platform_by_species_and_id( + conn: mdb.Connection, species_id: int, platformid: int +) -> Optional[dict]: + """Retrieve a platform by its species and ID""" + with conn.cursor(cursorclass=DictCursor) as cursor: + cursor.execute("SELECT * FROM GeneChip WHERE SpeciesId=%s AND Id=%s", + (species_id, platformid)) + result = cursor.fetchone()#pylint: disable=[duplicate-code] + if bool(result): + return dict(result) + + return None + + +def save_new_platform( + # pylint: disable=[too-many-arguments, too-many-positional-arguments] + cursor: Cursor, + species_id: int, + geo_platform: str, + platform_name: str, + platform_shortname: str, + platform_title: str, + go_tree_value: Optional[str] +) -> dict: + """Save a new platform to the database.""" + params = { + "species_id": species_id, + "GeoPlatform": geo_platform, + "GeneChipName": platform_name, + "Name": platform_shortname, + "Title": platform_title, + "GO_tree_value": go_tree_value + } + cursor.execute("SELECT SpeciesId, GeoPlatform FROM GeneChip") + assert (species_id, geo_platform) not in ( + (row["SpeciesId"], row["GeoPlatform"]) for row in cursor.fetchall()) + cursor.execute( + "INSERT INTO " + "GeneChip(SpeciesId, GeneChipName, Name, GeoPlatform, Title, GO_tree_value) " + "VALUES(" + "%(species_id)s, %(GeneChipName)s, %(Name)s, %(GeoPlatform)s, " + "%(Title)s, %(GO_tree_value)s" + ")", + params) + new_id = cursor.lastrowid + cursor.execute("UPDATE GeneChip SET GeneChipId=%s WHERE Id=%s", + (new_id, new_id)) + return { + **params, + "Id": new_id, + "GeneChipId": new_id + } diff --git a/uploader/platforms/views.py b/uploader/platforms/views.py new file mode 100644 index 0000000..d12a9ef --- /dev/null +++ b/uploader/platforms/views.py @@ -0,0 +1,118 @@ +"""The endpoints for the platforms""" +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.ui import make_template_renderer +from uploader.authorisation import require_login +from uploader.species.models import all_species, species_by_id +from uploader.datautils import safe_int, enumerate_sequence + +from .models import (save_new_platform, + platforms_by_species, + species_platforms_count) + +platformsbp = Blueprint("platforms", __name__) +render_template = make_template_renderer("platforms") + +@platformsbp.route("platforms", methods=["GET"]) +@require_login +def index(): + """Entry-point to the platforms feature.""" + with database_connection(app.config["SQL_URI"]) as conn: + if not bool(request.args.get("species_id")): + return render_template( + "platforms/index.html", + species=all_species(conn), + activelink="platforms") + + species_id = request.args.get("species_id") + if species_id == "CREATE-SPECIES": + return redirect(url_for( + "species.create_species", + return_to="species.platforms.list_platforms")) + + species = species_by_id(conn, request.args["species_id"]) + if not bool(species): + flash("No species selected.", "alert-danger") + return redirect(url_for("species.platforms.index")) + + return redirect(url_for("species.platforms.list_platforms", + species_id=species["SpeciesId"])) + + +@platformsbp.route("<int:species_id>/platforms", methods=["GET"]) +@require_login +def list_platforms(species_id: int): + """List all the available genetic sequencing platforms.""" + with database_connection(app.config["SQL_URI"]) as conn: + species = species_by_id(conn, species_id) + if not bool(species): + flash("No species provided.", "alert-danger") + return redirect(url_for("species.platforms.index")) + + 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( + "platforms/list-platforms.html", + species=species, + platforms=enumerate_sequence( + platforms_by_species(conn, + species_id, + offset=start_from, + limit=count), + start=start_from+1), + start_from=start_from, + count=count, + total_platforms=species_platforms_count(conn, species_id), + activelink="list-platforms") + + +@platformsbp.route("<int:species_id>/platforms/create", methods=["GET", "POST"]) +@require_login +def create_platform(species_id: int): + """Create a new genetic sequencing platform.""" + with (database_connection(app.config["SQL_URI"]) as conn, + conn.cursor(cursorclass=DictCursor) as cursor): + species = species_by_id(conn, species_id) + if not bool(species): + flash("No species provided.", "alert-danger") + return redirect(url_for("species.platforms.index")) + + if request.method == "GET": + return render_template( + "platforms/create-platform.html", + species=species, + activelink="create-platform") + + try: + form = request.form + _new_platform = save_new_platform( + cursor, + species_id, + form["geo-platform"], + form["platform-name"], + form["platform-shortname"], + form["platform-title"], + form.get("go-tree-value") or None) + except KeyError as _kerr: + flash(f"Required value for field {_kerr.args[0]} was not provided.", + "alert-danger") + return redirect(url_for("species.platforms.create_platform", + species_id=species_id)) + except AssertionError as _aerr: + flash(f"Platform with GeoPlatform value of '{form['geo-platform']}'" + f" already exists for species '{species['FullName']}'.", + "alert-danger") + return redirect(url_for("species.platforms.create_platform", + species_id=species_id)) + + flash("Platform created successfully", "alert-success") + return redirect(url_for("species.platforms.list_platforms", + species_id=species_id)) |