aboutsummaryrefslogtreecommitdiff
path: root/qc_app/samples.py
diff options
context:
space:
mode:
Diffstat (limited to 'qc_app/samples.py')
-rw-r--r--qc_app/samples.py120
1 files changed, 61 insertions, 59 deletions
diff --git a/qc_app/samples.py b/qc_app/samples.py
index 1063fb8..dee08e5 100644
--- a/qc_app/samples.py
+++ b/qc_app/samples.py
@@ -17,8 +17,8 @@ from flask import (
from quality_control.parsing import take
from .files import save_file
-from .db_utils import with_db_connection
from .dbinsert import species_by_id, groups_by_species
+from .db_utils import with_db_connection, database_connection
samples = Blueprint("samples", __name__)
@@ -29,7 +29,8 @@ def select_species():
species_id = request.form.get("species_id")
if bool(species_id):
species_id = int(species_id)
- species = species_by_id(species_id)
+ species = with_db_connection(
+ lambda conn: species_by_id(conn, species_id))
if bool(species):
return render_template(
"samples/select-population.html",
@@ -72,40 +73,42 @@ def population_by_id(conn: mdb.Connection, population_id: int) -> dict:
def create_population():
"""Create new grouping/population."""
species_page = redirect(url_for("samples.select_species"), code=307)
- species = species_by_id(request.form.get("species_id"))
- pop_name = request.form.get("inbredset_name").strip()
- pop_fullname = request.form.get("inbredset_fullname").strip()
-
- if not bool(species):
- flash("Invalid species!", "alert-error error-create-population")
- return species_page
- if (not bool(pop_name)) or (not bool(pop_fullname)):
- flash("You *MUST* provide a grouping/population name",
- "alert-error error-create-population")
- return species_page
-
- pop_id = with_db_connection(lambda conn: save_population(conn, {
- "SpeciesId": species["SpeciesId"],
- "Name": pop_name,
- "InbredSetName": pop_fullname,
- "FullName": pop_fullname,
- "Family": request.form.get("inbredset_family") or None,
- "Description": request.form.get("description") or None
- }))
- flash("Grouping/Population created successfully.", "alert-success")
- return render_template(
- "samples/upload-samples.html",
- species=species,
- population=with_db_connection(
- lambda conn: population_by_id(conn, pop_id)))
+ with database_connection(app.config["SQL_URI"]) as conn:
+ species = species_by_id(conn, request.form.get("species_id"))
+ pop_name = request.form.get("inbredset_name").strip()
+ pop_fullname = request.form.get("inbredset_fullname").strip()
+
+ if not bool(species):
+ flash("Invalid species!", "alert-error error-create-population")
+ return species_page
+ if (not bool(pop_name)) or (not bool(pop_fullname)):
+ flash("You *MUST* provide a grouping/population name",
+ "alert-error error-create-population")
+ return species_page
+
+ pop_id = save_population(conn, {
+ "SpeciesId": species["SpeciesId"],
+ "Name": pop_name,
+ "InbredSetName": pop_fullname,
+ "FullName": pop_fullname,
+ "Family": request.form.get("inbredset_family") or None,
+ "Description": request.form.get("description") or None
+ })
+ flash("Grouping/Population created successfully.", "alert-success")
+ return render_template(
+ "samples/upload-samples.html",
+ species=species,
+ population=with_db_connection(
+ lambda conn: population_by_id(conn, pop_id)))
@samples.route("/upload/select-population", methods=["POST"])
def select_population():
"""Select from existing groupings/populations."""
species_page = redirect(url_for("samples.select_species"), code=307)
- species = species_by_id(request.form.get("species_id"))
- pop_id = int(request.form.get("inbredset_id"))
- population = with_db_connection(lambda conn: population_by_id(conn, pop_id))
+ with database_connection(app.config["SQL_URI"]) as conn:
+ species = species_by_id(conn, request.form.get("species_id"))
+ pop_id = int(request.form.get("inbredset_id"))
+ population = with_db_connection(lambda conn: population_by_id(conn, pop_id))
if not bool(species):
flash("Invalid species!", "alert-error error-select-population")
@@ -195,33 +198,33 @@ def upload_samples():
samples_uploads_page = redirect(url_for("samples.select_population"),
code=307)
- species = species_by_id(request.form.get("species_id"))
- if not bool(species):
- flash("Invalid species!", "alert-error")
- return samples_uploads_page
-
- population = with_db_connection(
- lambda conn: population_by_id(
- conn, int(request.form.get("inbredset_id"))))
- if not bool(population):
- flash("Invalid grouping/population!", "alert-error")
- return samples_uploads_page
-
- samples_file = save_file(request.files["samples_file"], Path(app.config["UPLOAD_FOLDER"]))
- if not bool(samples_file):
- flash("You need to provide a file with the samples data.")
- return samples_uploads_page
-
- firstlineheading = (request.form.get("first_line_heading") == "on")
-
- separator = request.form.get("separator")
- if separator == "other":
- separator = request.form.get("other_separator")
- if not bool(separator):
- flash("You need to provide a separator character.", "alert-error")
- return samples_uploads_page
+ with database_connection(app.config["SQL_URI"]) as conn:
+ species = species_by_id(conn, request.form.get("species_id"))
+ if not bool(species):
+ flash("Invalid species!", "alert-error")
+ return samples_uploads_page
+
+ population = with_db_connection(
+ lambda conn: population_by_id(
+ conn, int(request.form.get("inbredset_id"))))
+ if not bool(population):
+ flash("Invalid grouping/population!", "alert-error")
+ return samples_uploads_page
+
+ samples_file = save_file(request.files["samples_file"], Path(app.config["UPLOAD_FOLDER"]))
+ if not bool(samples_file):
+ flash("You need to provide a file with the samples data.")
+ return samples_uploads_page
+
+ firstlineheading = (request.form.get("first_line_heading") == "on")
+
+ separator = request.form.get("separator")
+ if separator == "other":
+ separator = request.form.get("other_separator")
+ if not bool(separator):
+ flash("You need to provide a separator character.", "alert-error")
+ return samples_uploads_page
- def __insert_samples__(conn: mdb.Connection):
save_samples_data(
conn,
species["SpeciesId"],
@@ -232,5 +235,4 @@ def upload_samples():
population["InbredSetId"],
(row["Name"] for row in read_samples_file(samples_file, separator, firstlineheading)))
- with_db_connection(__insert_samples__)
- return "SUCCESS: Respond with a better UI than this."
+ return "SUCCESS: Respond with a better UI than this."