diff options
| -rw-r--r-- | qc_app/db/__init__.py | 7 | ||||
| -rw-r--r-- | qc_app/db/populations.py | 49 | ||||
| -rw-r--r-- | qc_app/db/species.py | 24 | ||||
| -rw-r--r-- | qc_app/dbinsert.py | 33 | ||||
| -rw-r--r-- | qc_app/samples.py | 29 |
5 files changed, 80 insertions, 62 deletions
diff --git a/qc_app/db/__init__.py b/qc_app/db/__init__.py new file mode 100644 index 0000000..0b48461 --- /dev/null +++ b/qc_app/db/__init__.py @@ -0,0 +1,7 @@ +"""Database functions""" +from .species import species, species_by_id +from .populations import ( + save_population, + population_by_id, + populations_by_species, + population_by_species_and_id) diff --git a/qc_app/db/populations.py b/qc_app/db/populations.py new file mode 100644 index 0000000..0bcbe5d --- /dev/null +++ b/qc_app/db/populations.py @@ -0,0 +1,49 @@ +"""Functions for accessing the database relating to species populations.""" +import MySQLdb as mdb +from MySQLdb.cursors import DictCursor + +def population_by_id(conn: mdb.Connection, population_id) -> dict: + """Get the grouping/population by id.""" + with conn.cursor(cursorclass=DictCursor) as cursor: + cursor.execute("SELECT * FROM InbredSet WHERE InbredSetId=%s", + (population_id,)) + return cursor.fetchone() + +def population_by_species_and_id( + conn: mdb.Connection, species_id, population_id) -> dict: + """Retrieve a population by its identifier and species.""" + with conn.cursor(cursorclass=DictCursor) as cursor: + cursor.execute("SELECT * FROM InbredSet WHERE SpeciesId=%s AND Id=%s", + (species_id, population_id)) + return cursor.fetchone() + +def populations_by_species(conn: mdb.Connection, speciesid) -> tuple: + "Retrieve group (InbredSet) information from the database." + with conn.cursor(cursorclass=DictCursor) as cursor: + query = "SELECT * FROM InbredSet WHERE SpeciesId=%s" + cursor.execute(query, (speciesid,)) + return tuple(cursor.fetchall()) + + return tuple() + +def save_population(conn: mdb.Connection, population_details: dict) -> dict: + """Save the population details to the db.""" + with conn.cursor() as cursor: + cursor.execute("SELECT MAX(Id) AS last_id FROM InbredSet") + new_id = cursor.fetchone()["last_id"] + 1 + cursor.execute( + "INSERT INTO InbredSet(" + "Id, InbredSetId, InbredSetName, Name, SpeciesId, FullName, " + "MenuOrderId, Description" + ") " + "VALUES (" + "%(Id)s, %(InbredSetId)s, %(InbredSetName)s, %(Name)s, " + "%(SpeciesId)s, %(FullName)s, %(MenuOrderId)s, %(Description)s" + ")", + { + "Id": new_id, + "InbredSetId": new_id, + "MenuOrderId": 0, + **population_details + }) + return {**population_details, "population_id": new_id} diff --git a/qc_app/db/species.py b/qc_app/db/species.py new file mode 100644 index 0000000..7006d15 --- /dev/null +++ b/qc_app/db/species.py @@ -0,0 +1,24 @@ +"""Database functions for species.""" +import MySQLdb as mdb +from MySQLdb.cursors import DictCursor + +def species(conn: mdb.Connection) -> tuple: + "Retrieve the species from the database." + with conn.cursor(cursorclass=DictCursor) as cursor: + cursor.execute( + "SELECT SpeciesId, SpeciesName, LOWER(Name) AS Name, MenuName " + "FROM Species") + return tuple(cursor.fetchall()) + + return tuple() + +def species_by_id(conn: mdb.Connection, speciesid) -> dict: + "Retrieve the species from the database by id." + with conn.cursor(cursorclass=DictCursor) as cursor: + cursor.execute( + ( + "SELECT " + "SpeciesId, SpeciesName, LOWER(Name) AS Name, MenuName " + "FROM Species WHERE SpeciesId=%s"), + (speciesid,)) + return cursor.fetchone() diff --git a/qc_app/dbinsert.py b/qc_app/dbinsert.py index 2282c8d..881c913 100644 --- a/qc_app/dbinsert.py +++ b/qc_app/dbinsert.py @@ -5,7 +5,6 @@ from typing import Union from functools import reduce from datetime import datetime -import MySQLdb as mdb from redis import Redis from MySQLdb.cursors import DictCursor from flask import ( @@ -31,28 +30,6 @@ def make_menu_items_grouper(grouping_fn=lambda item: item): return {**acc, grouping: (acc[grouping] + (row_values,))} return __grouper__ -def species() -> tuple: - "Retrieve the species from the database." - with database_connection() as conn: - with conn.cursor(cursorclass=DictCursor) as cursor: - cursor.execute( - "SELECT SpeciesId, SpeciesName, LOWER(Name) AS Name, MenuName " - "FROM Species") - return tuple(cursor.fetchall()) - - return tuple() - -def species_by_id(conn: mdb.Connection, speciesid) -> Union[dict, None]: - "Retrieve the species from the database by id." - with conn.cursor(cursorclass=DictCursor) as cursor: - cursor.execute( - ( - "SELECT " - "SpeciesId, SpeciesName, LOWER(Name) AS Name, MenuName " - "FROM Species WHERE SpeciesId=%s"), - (speciesid,)) - return cursor.fetchone() - def genechips(): "Retrieve the genechip information from the database" def __organise_by_species__(acc, chip): @@ -94,16 +71,6 @@ def studies_by_species_and_platform(speciesid:int, genechipid:int) -> tuple: return tuple() -def groups_by_species(speciesid:int) -> tuple: - "Retrieve group (InbredSet) information from the database." - with database_connection() as conn: - with conn.cursor(cursorclass=DictCursor) as cursor: - query = "SELECT * FROM InbredSet WHERE SpeciesId=%s" - cursor.execute(query, (speciesid,)) - return tuple(cursor.fetchall()) - - return tuple() - def organise_groups_by_family(acc:dict, group:dict) -> dict: "Organise the group (InbredSet) information by the group field" family = group["Family"] diff --git a/qc_app/samples.py b/qc_app/samples.py index d27164e..107e889 100644 --- a/qc_app/samples.py +++ b/qc_app/samples.py @@ -48,35 +48,6 @@ def select_species(): flash("You need to select a species", "alert-error") return index_page -def save_population(conn: mdb.Connection, population_details: dict) -> int: - """Save the population details to the db.""" - with conn.cursor(cursorclass=DictCursor) as cursor: - cursor.execute("SELECT MAX(Id) AS last_id FROM InbredSet") - new_id = cursor.fetchone()["last_id"] + 1 - cursor.execute( - "INSERT INTO InbredSet(" - "Id, InbredSetId, InbredSetName, Name, SpeciesId, FullName, " - "MenuOrderId, Description" - ") " - "VALUES (" - "%(Id)s, %(InbredSetId)s, %(InbredSetName)s, %(Name)s, " - "%(SpeciesId)s, %(FullName)s, %(MenuOrderId)s, %(Description)s" - ")", - { - "Id": new_id, - "InbredSetId": new_id, - "MenuOrderId": 0, - **population_details - }) - return new_id - -def population_by_id(conn: mdb.Connection, population_id: int) -> dict: - """Get the grouping/population by id.""" - with conn.cursor(cursorclass=DictCursor) as cursor: - cursor.execute("SELECT * FROM InbredSet WHERE InbredSetId=%s", - (population_id,)) - return cursor.fetchone() - @samples.route("/upload/create-population", methods=["POST"]) def create_population(): """Create new grouping/population.""" |