aboutsummaryrefslogtreecommitdiff
path: root/qc_app
diff options
context:
space:
mode:
Diffstat (limited to 'qc_app')
-rw-r--r--qc_app/db/__init__.py7
-rw-r--r--qc_app/db/populations.py49
-rw-r--r--qc_app/db/species.py24
-rw-r--r--qc_app/dbinsert.py33
-rw-r--r--qc_app/samples.py29
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."""
generated by cgit v1.2.3 (git 2.44.0) at 2024-11-10 15:45:41 +0000