aboutsummaryrefslogtreecommitdiff
path: root/qc_app/db
diff options
context:
space:
mode:
Diffstat (limited to 'qc_app/db')
-rw-r--r--qc_app/db/__init__.py8
-rw-r--r--qc_app/db/averaging.py23
-rw-r--r--qc_app/db/datasets.py133
-rw-r--r--qc_app/db/platforms.py25
-rw-r--r--qc_app/db/populations.py54
-rw-r--r--qc_app/db/species.py22
-rw-r--r--qc_app/db/tissues.py50
7 files changed, 0 insertions, 315 deletions
diff --git a/qc_app/db/__init__.py b/qc_app/db/__init__.py
deleted file mode 100644
index 36e93e8..0000000
--- a/qc_app/db/__init__.py
+++ /dev/null
@@ -1,8 +0,0 @@
-"""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)
-from .datasets import geno_datasets_by_species_and_population
diff --git a/qc_app/db/averaging.py b/qc_app/db/averaging.py
deleted file mode 100644
index 62bbe67..0000000
--- a/qc_app/db/averaging.py
+++ /dev/null
@@ -1,23 +0,0 @@
-"""Functions for db interactions for averaging methods"""
-from typing import Optional
-
-import MySQLdb as mdb
-from MySQLdb.cursors import DictCursor
-
-def averaging_methods(conn: mdb.Connection) -> tuple[dict, ...]:
- """Fetch all available averaging methods"""
- with conn.cursor(cursorclass=DictCursor) as cursor:
- cursor.execute("SELECT * FROM AvgMethod")
- return tuple(dict(row) for row in cursor.fetchall())
-
-def averaging_method_by_id(
- conn: mdb.Connection, averageid: int) -> Optional[dict]:
- """Fetch the averaging method by its ID"""
- with conn.cursor(cursorclass=DictCursor) as cursor:
- cursor.execute("SELECT * FROM AvgMethod WHERE Id=%s",
- (averageid,))
- result = cursor.fetchone()
- if bool(result):
- return dict(result)
-
- return None
diff --git a/qc_app/db/datasets.py b/qc_app/db/datasets.py
deleted file mode 100644
index 767ec41..0000000
--- a/qc_app/db/datasets.py
+++ /dev/null
@@ -1,133 +0,0 @@
-"""Functions for accessing the database relating to datasets."""
-from datetime import date
-from typing import Optional
-
-import MySQLdb as mdb
-from MySQLdb.cursors import DictCursor
-
-def geno_datasets_by_species_and_population(
- conn: mdb.Connection,
- speciesid: int,
- populationid: int) -> tuple[dict, ...]:
- """Retrieve all genotypes datasets by species and population"""
- with conn.cursor(cursorclass=DictCursor) as cursor:
- cursor.execute(
- "SELECT gf.* FROM InbredSet AS iset INNER JOIN GenoFreeze AS gf "
- "ON iset.InbredSetId=gf.InbredSetId "
- "WHERE iset.SpeciesId=%(sid)s AND iset.InbredSetId=%(pid)s",
- {"sid": speciesid, "pid": populationid})
- return tuple(dict(row) for row in cursor.fetchall())
-
-def geno_dataset_by_id(conn: mdb.Connection, dataset_id) -> Optional[dict]:
- """Retrieve genotype dataset by ID"""
- with conn.cursor(cursorclass=DictCursor) as cursor:
- cursor.execute("SELECT * FROM GenoFreeze WHERE Id=%s", (dataset_id,))
- _dataset = cursor.fetchone()
- return dict(_dataset) if bool(_dataset) else None
-
-def probeset_studies_by_species_and_population(
- conn: mdb.Connection,
- speciesid: int,
- populationid: int) -> tuple[dict, ...]:
- """Retrieve all probesets"""
- with conn.cursor(cursorclass=DictCursor) as cursor:
- cursor.execute(
- "SELECT pf.* FROM InbredSet AS iset INNER JOIN ProbeFreeze AS pf "
- "ON iset.InbredSetId=pf.InbredSetId "
- "WHERE iset.SpeciesId=%(sid)s AND iset.InbredSetId=%(pid)s",
- {"sid": speciesid, "pid": populationid})
- return tuple(dict(row) for row in cursor.fetchall())
-
-def probeset_datasets_by_study(conn: mdb.Connection,
- studyid: int) -> tuple[dict, ...]:
- """Retrieve all probeset databases by study."""
- with conn.cursor(cursorclass=DictCursor) as cursor:
- cursor.execute("SELECT * FROM ProbeSetFreeze WHERE ProbeFreezeId=%s",
- (studyid,))
- return tuple(dict(row) for row in cursor.fetchall())
-
-def probeset_study_by_id(conn: mdb.Connection, studyid) -> Optional[dict]:
- """Retrieve ProbeSet study by ID"""
- with conn.cursor(cursorclass=DictCursor) as cursor:
- cursor.execute("SELECT * FROM ProbeFreeze WHERE Id=%s", (studyid,))
- _study = cursor.fetchone()
- return dict(_study) if bool(_study) else None
-
-def probeset_create_study(conn: mdb.Connection,#pylint: disable=[too-many-arguments]
- populationid: int,
- platformid: int,
- tissueid: int,
- studyname: str,
- studyfullname: str = "",
- studyshortname: str = ""):
- """Create a new ProbeSet study."""
- with conn.cursor(cursorclass=DictCursor) as cursor:
- studydata = {
- "platid": platformid,
- "tissueid": tissueid,
- "name": studyname,
- "fname": studyfullname or studyname,
- "sname": studyshortname,
- "today": date.today().isoformat(),
- "popid": populationid
- }
- cursor.execute(
- """
- INSERT INTO ProbeFreeze(
- ChipId, TissueId, Name, FullName, ShortName, CreateTime,
- InbredSetId
- ) VALUES (
- %(platid)s, %(tissueid)s, %(name)s, %(fname)s, %(sname)s,
- %(today)s, %(popid)s
- )
- """,
- studydata)
- studyid = cursor.lastrowid
- cursor.execute("UPDATE ProbeFreeze SET ProbeFreezeId=%s WHERE Id=%s",
- (studyid, studyid))
- return {**studydata, "studyid": studyid}
-
-def probeset_create_dataset(conn: mdb.Connection,#pylint: disable=[too-many-arguments]
- studyid: int,
- averageid: int,
- datasetname: str,
- datasetfullname: str,
- datasetshortname: str="",
- public: bool = True,
- datascale="log2") -> dict:
- """Create a new ProbeSet dataset."""
- with conn.cursor(cursorclass=DictCursor) as cursor:
- dataset = {
- "studyid": studyid,
- "averageid": averageid,
- "name2": datasetname,
- "fname": datasetfullname,
- "name": datasetshortname,
- "sname": datasetshortname,
- "today": date.today().isoformat(),
- "public": 2 if public else 0,
- "authorisedusers": "williamslab",
- "datascale": datascale
- }
- cursor.execute(
- """
- INSERT INTO ProbeSetFreeze(
- ProbeFreezeId, AvgId, Name, Name2, FullName, ShortName,
- CreateTime, public, AuthorisedUsers, DataScale)
- VALUES(
- %(studyid)s, %(averageid)s, %(name)s, %(name2)s, %(fname)s,
- %(sname)s, %(today)s, %(public)s, %(authorisedusers)s,
- %(datascale)s)
- """,
- dataset)
- return {**dataset, "datasetid": cursor.lastrowid}
-
-def probeset_dataset_by_id(conn: mdb.Connection, datasetid) -> Optional[dict]:
- """Fetch a ProbeSet dataset by its ID"""
- with conn.cursor(cursorclass=DictCursor) as cursor:
- cursor.execute("SELECT * FROM ProbeSetFreeze WHERE Id=%s", (datasetid,))
- result = cursor.fetchone()
- if bool(result):
- return dict(result)
-
- return None
diff --git a/qc_app/db/platforms.py b/qc_app/db/platforms.py
deleted file mode 100644
index cb527a7..0000000
--- a/qc_app/db/platforms.py
+++ /dev/null
@@ -1,25 +0,0 @@
-"""Handle db interactions for platforms."""
-from typing import Optional
-
-import MySQLdb as mdb
-from MySQLdb.cursors import DictCursor
-
-def platforms_by_species(
- conn: mdb.Connection, speciesid: int) -> tuple[dict, ...]:
- """Retrieve platforms by the species"""
- with conn.cursor(cursorclass=DictCursor) as cursor:
- cursor.execute("SELECT * FROM GeneChip WHERE SpeciesId=%s "
- "ORDER BY GeneChipName ASC",
- (speciesid,))
- return tuple(dict(row) for row in cursor.fetchall())
-
-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
diff --git a/qc_app/db/populations.py b/qc_app/db/populations.py
deleted file mode 100644
index 4485e52..0000000
--- a/qc_app/db/populations.py
+++ /dev/null
@@ -1,54 +0,0 @@
-"""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(cursorclass=DictCursor) as cursor:
- cursor.execute(
- "INSERT INTO InbredSet("
- "InbredSetId, InbredSetName, Name, SpeciesId, FullName, "
- "MenuOrderId, Description"
- ") "
- "VALUES ("
- "%(InbredSetId)s, %(InbredSetName)s, %(Name)s, %(SpeciesId)s, "
- "%(FullName)s, %(MenuOrderId)s, %(Description)s"
- ")",
- {
- "MenuOrderId": 0,
- "InbredSetId": 0,
- **population_details
- })
- new_id = cursor.lastrowid
- cursor.execute("UPDATE InbredSet SET InbredSetId=%s WHERE Id=%s",
- (new_id, new_id))
- return {
- **population_details,
- "Id": new_id,
- "InbredSetId": new_id,
- "population_id": new_id
- }
diff --git a/qc_app/db/species.py b/qc_app/db/species.py
deleted file mode 100644
index 653e59b..0000000
--- a/qc_app/db/species.py
+++ /dev/null
@@ -1,22 +0,0 @@
-"""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, "
- "FullName 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, "
- "FullName FROM Species WHERE SpeciesId=%s",
- (speciesid,))
- return cursor.fetchone()
diff --git a/qc_app/db/tissues.py b/qc_app/db/tissues.py
deleted file mode 100644
index 9fe7bab..0000000
--- a/qc_app/db/tissues.py
+++ /dev/null
@@ -1,50 +0,0 @@
-"""Handle db interactions for tissue."""
-from typing import Union, Optional
-
-import MySQLdb as mdb
-from MySQLdb.cursors import DictCursor
-
-def all_tissues(conn: mdb.Connection) -> tuple[dict, ...]:
- """All available tissue."""
- with conn.cursor(cursorclass=DictCursor) as cursor:
- cursor.execute("SELECT * FROM Tissue ORDER BY TissueName")
- return tuple(dict(row) for row in cursor.fetchall())
-
-
-def tissue_by_id(conn: mdb.Connection, tissueid) -> Optional[dict]:
- """Retrieve a tissue by its ID"""
- with conn.cursor(cursorclass=DictCursor) as cursor:
- cursor.execute("SELECT * FROM Tissue WHERE Id=%s", (tissueid,))
- result = cursor.fetchone()
- if bool(result):
- return dict(result)
-
- return None
-
-
-def create_new_tissue(
- conn: mdb.Connection,
- name: str,
- shortname: str,
- birnlexid: Optional[str] = None,
- birnlexname: Optional[str] = None
-) -> dict[str, Union[int, str, None]]:
- """Add a new tissue, organ or biological material to the database."""
- with conn.cursor() as cursor:
- cursor.execute(
- "INSERT INTO "
- "Tissue(TissueName, Name, Short_Name, BIRN_lex_ID, BIRN_lex_Name) "
- "VALUES (%s, %s, %s, %s, %s)",
- (name, name, shortname, birnlexid, birnlexname))
- tissueid = cursor.lastrowid
- cursor.execute("UPDATE Tissue SET TissueId=%s WHERE Id=%s",
- (tissueid, tissueid))
- return {
- "Id": tissueid,
- "TissueId": tissueid,
- "TissueName": name,
- "Name": name,
- "Short_Name": shortname,
- "BIRN_lex_ID": birnlexid,
- "BIRN_lex_Name": birnlexname
- }
generated by cgit v1.2.3 (git 2.44.0) at 2024-11-10 11:39:55 +0000