landon/Submissions-App
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update to autofill.

Browse Source
This commit is contained in:
Landon Wark
2023-09-20 09:19:59 -05:00
parent 0c843d1561
commit 82ab06efad
13 changed files with 378 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files

142
src/submissions/backend/db/functions.py
View File

@@ -18,7 +18,7 @@ import numpy as np
import yaml
from pathlib import Path
from tools import Settings, check_regex_match, RSLNamer
from typing import List
from typing import List, Tuple
@@ -27,6 +27,14 @@ logger = logging.getLogger(f"submissions.{__name__}")
# The below _should_ allow automatic creation of foreign keys in the database
@event.listens_for(Engine, "connect")
def set_sqlite_pragma(dbapi_connection, connection_record):
"""
*should* allow automatic creation of foreign keys in the database
I have no idea how it actually works.
Args:
dbapi_connection (_type_): _description_
connection_record (_type_): _description_
"""
cursor = dbapi_connection.cursor()
cursor.execute("PRAGMA foreign_keys=ON")
cursor.close()
@@ -79,7 +87,7 @@ def store_reagent(ctx:Settings, reagent:models.Reagent) -> None|dict:
return {"message":"The database is locked for editing."}
return None
def construct_submission_info(ctx:Settings, info_dict:dict) -> models.BasicSubmission:
def construct_submission_info(ctx:Settings, info_dict:dict) -> Tuple[models.BasicSubmission, dict]:
"""
Construct submission object from dictionary
@@ -273,7 +281,7 @@ def lookup_reagenttype_by_name(ctx:Settings, rt_name:str) -> models.ReagentType:
Returns:
models.ReagentType: looked up reagent type
"""
logger.debug(f"Looking up ReagentType by name: {rt_name.title()}")
logger.debug(f"Looking up ReagentType by name: {rt_name}")
lookedup = ctx.database_session.query(models.ReagentType).filter(models.ReagentType.name==rt_name).first()
logger.debug(f"Found ReagentType: {lookedup}")
return lookedup
@@ -302,7 +310,7 @@ def lookup_kittype_by_name(ctx:Settings, name:str|dict) -> models.KitType:
Args:
ctx (Settings): settings object passed from bui
name (str): name of kit to query
name (str|dict): name of kit to query, or parsed object containing value=name
Returns:
models.KitType: retrieved kittype
@@ -989,25 +997,6 @@ def lookup_reagent(ctx:Settings, reagent_lot:str, type_name:str|None=None) -> mo
# return ctx['database_session'].query(models.Reagent).filter(models.Reagent.lot==reagent_lot).first()
return ctx.database_session.query(models.Reagent).filter(models.Reagent.lot==reagent_lot).first()
def lookup_last_used_reagenttype_lot(ctx:Settings, type_name:str) -> models.Reagent:
"""
Look up the last used reagent of the reagent type
Args:
ctx (Settings): Settings object passed down from gui
type_name (str): Name of reagent type
Returns:
models.Reagent: Reagent object with last used lot.
"""
# rt = ctx['database_session'].query(models.ReagentType).filter(models.ReagentType.name==type_name).first()
rt = ctx.database_session.query(models.ReagentType).filter(models.ReagentType.name==type_name).first()
logger.debug(f"Reagent type looked up for {type_name}: {rt.__str__()}")
try:
return lookup_reagent(ctx=ctx, reagent_lot=rt.last_used, type_name=type_name)
except AttributeError:
return None
def check_kit_integrity(sub:models.BasicSubmission|models.KitType, reagenttypes:list|None=None) -> dict|None:
"""
Ensures all reagents expected in kit are listed in Submission
@@ -1120,7 +1109,7 @@ def lookup_subsamp_association_by_plate_sample(ctx:Settings, rsl_plate_num:str,
.filter(models.BasicSample.submitter_id==rsl_sample_num)\
.first()
def lookup_sub_wwsamp_association_by_plate_sample(ctx:Settings, rsl_plate_num:str, rsl_sample_num:str) -> models.WastewaterAssociation:
def lookup_sub_samp_association_by_plate_sample(ctx:Settings, rsl_plate_num:str|models.BasicSample, rsl_sample_num:str|models.BasicSubmission) -> models.WastewaterAssociation:
"""
_summary_
@@ -1132,12 +1121,36 @@ def lookup_sub_wwsamp_association_by_plate_sample(ctx:Settings, rsl_plate_num:st
Returns:
models.SubmissionSampleAssociation: _description_
"""
return ctx.database_session.query(models.WastewaterAssociation)\
.join(models.Wastewater)\
.join(models.WastewaterSample)\
.filter(models.BasicSubmission.rsl_plate_num==rsl_plate_num)\
.filter(models.BasicSample.submitter_id==rsl_sample_num)\
.first()
# logger.debug(f"{type(rsl_plate_num)}, {type(rsl_sample_num)}")
match rsl_plate_num:
case models.BasicSubmission()|models.Wastewater():
# logger.debug(f"Model for rsl_plate_num: {rsl_plate_num}")
first_query = ctx.database_session.query(models.SubmissionSampleAssociation)\
.filter(models.SubmissionSampleAssociation.submission==rsl_plate_num)
case str():
# logger.debug(f"String for rsl_plate_num: {rsl_plate_num}")
first_query = ctx.database_session.query(models.SubmissionSampleAssociation)\
.join(models.BasicSubmission)\
.filter(models.BasicSubmission.rsl_plate_num==rsl_plate_num)
case _:
logger.error(f"Unknown case for rsl_plate_num {rsl_plate_num}")
match rsl_sample_num:
case models.BasicSample()|models.WastewaterSample():
# logger.debug(f"Model for rsl_sample_num: {rsl_sample_num}")
second_query = first_query.filter(models.SubmissionSampleAssociation.sample==rsl_sample_num)
# case models.WastewaterSample:
# second_query = first_query.filter(models.SubmissionSampleAssociation.sample==rsl_sample_num)
case str():
# logger.debug(f"String for rsl_sample_num: {rsl_sample_num}")
second_query = first_query.join(models.BasicSample)\
.filter(models.BasicSample.submitter_id==rsl_sample_num)
case _:
logger.error(f"Unknown case for rsl_sample_num {rsl_sample_num}")
try:
return second_query.first()
except UnboundLocalError:
logger.error(f"Couldn't construct second query")
return None
def lookup_all_reagent_names_by_role(ctx:Settings, role_name:str) -> List[str]:
"""
@@ -1183,7 +1196,7 @@ def add_reagenttype_to_kit(ctx:Settings, rt_name:str, kit_name:str, eol:int=0):
kit = lookup_kittype_by_name(ctx=ctx, name=kit_name)
rt = lookup_reagenttype_by_name(ctx=ctx, rt_name=rt_name)
if rt == None:
rt = models.ReagentType(name=rt_name.strip(), eol_ext=timedelta(30*eol), last_used="")
rt = models.ReagentType(name=rt_name.strip(), eol_ext=timedelta(30*eol))
ctx.database_session.add(rt)
assoc = models.KitTypeReagentTypeAssociation(kit_type=kit, reagent_type=rt, uses={})
kit.kit_reagenttype_associations.append(assoc)
@@ -1203,4 +1216,69 @@ def update_subsampassoc_with_pcr(ctx:Settings, submission:models.BasicSubmission
except AttributeError:
logger.error(f"Can't set {k} to {v}")
ctx.database_session.add(assoc)
ctx.database_session.commit()
ctx.database_session.commit()
def lookup_ww_sample_by_processing_number(ctx:Settings, processing_number:str):
return ctx.database_session.query(models.WastewaterSample).filter(models.WastewaterSample.ww_processing_num==processing_number).first()
def lookup_kitreagentassoc_by_kit_and_reagent(ctx:Settings, kit:models.KitType|str, reagent_type:models.ReagentType|str) -> models.KitTypeReagentTypeAssociation:
"""
_summary_
Args:
ctx (Settings): _description_
kit (models.KitType | str): _description_
reagent_type (models.ReagentType | str): _description_
Returns:
models.KitTypeReagentTypeAssociation: _description_
"""
base_query = ctx.database_session.query(models.KitTypeReagentTypeAssociation)
match kit:
case models.KitType():
query1 = base_query.filter(models.KitTypeReagentTypeAssociation.kit_type==kit)
case str():
query1 = base_query.join(models.KitType).filter(models.KitType.name==kit)
case _:
query1 = base_query
match reagent_type:
case models.ReagentType():
query2 = query1.filter(models.KitTypeReagentTypeAssociation.reagent_type==reagent_type)
case str():
query2 = query1.join(models.ReagentType).filter(models.ReagentType.name==reagent_type)
case _:
query2 = query1
return query2.first()
def lookup_last_used_reagenttype_lot(ctx:Settings, type_name:str, extraction_kit:str|None=None) -> models.Reagent:
"""
Look up the last used reagent of the reagent type
Args:
ctx (Settings): Settings object passed down from gui
type_name (str): Name of reagent type
Returns:
models.Reagent: Reagent object with last used lot.
"""
assoc = lookup_kitreagentassoc_by_kit_and_reagent(ctx=ctx, kit=extraction_kit, reagent_type=type_name)
return lookup_reagent(ctx=ctx, reagent_lot=assoc.last_used)
def update_last_used(ctx:Settings, reagent:models.Reagent, kit:models.KitType):
"""
_summary_
Args:
ctx (Settings): _description_
reagent (models.ReagentType): _description_
reagent_lot (str): _description_
"""
rt = list(set(reagent.type).intersection(kit.reagent_types))[0]
if rt != None:
assoc = lookup_kitreagentassoc_by_kit_and_reagent(ctx=ctx, kit=kit, reagent_type=rt)
if assoc != None:
if assoc.last_used != reagent.lot:
logger.debug(f"Updating {assoc} last used to {reagent.lot}")
assoc.last_used = reagent.lot
ctx.database_session.merge(assoc)
ctx.database_session.commit()

86
src/submissions/backend/db/models/kits.py
View File

@@ -97,39 +97,7 @@ class KitType(Base):
map['info'] = {}
return map
class KitTypeReagentTypeAssociation(Base):
"""
table containing reagenttype/kittype associations
DOC: https://docs.sqlalchemy.org/en/14/orm/extensions/associationproxy.html
"""
__tablename__ = "_reagenttypes_kittypes"
reagent_types_id = Column(INTEGER, ForeignKey("_reagent_types.id"), primary_key=True)
kits_id = Column(INTEGER, ForeignKey("_kits.id"), primary_key=True)
uses = Column(JSON)
required = Column(INTEGER)
kit_type = relationship(KitType, back_populates="kit_reagenttype_associations")
# reference to the "ReagentType" object
reagent_type = relationship("ReagentType")
def __init__(self, kit_type=None, reagent_type=None, uses=None, required=1):
self.kit_type = kit_type
self.reagent_type = reagent_type
self.uses = uses
self.required = required
@validates('required')
def validate_age(self, key, value):
if not 0 <= value < 2:
raise ValueError(f'Invalid required value {value}. Must be 0 or 1.')
return value
@validates('reagenttype')
def validate_reagenttype(self, key, value):
if not isinstance(value, ReagentType):
raise ValueError(f'{value} is not a reagenttype')
return value
class ReagentType(Base):
"""
@@ -141,7 +109,16 @@ class ReagentType(Base):
name = Column(String(64)) #: name of reagent type
instances = relationship("Reagent", back_populates="type", secondary=reagenttypes_reagents) #: concrete instances of this reagent type
eol_ext = Column(Interval()) #: extension of life interval
last_used = Column(String(32)) #: last used lot number of this type of reagent
reagenttype_kit_associations = relationship(
"KitTypeReagentTypeAssociation",
back_populates="reagent_type",
cascade="all, delete-orphan",
)
# association proxy of "user_keyword_associations" collection
# to "keyword" attribute
kit_types = association_proxy("kit_reagenttype_associations", "kit_type")
@validates('required')
def validate_age(self, key, value):
@@ -160,6 +137,44 @@ class ReagentType(Base):
def __repr__(self):
return f"ReagentType({self.name})"
class KitTypeReagentTypeAssociation(Base):
"""
table containing reagenttype/kittype associations
DOC: https://docs.sqlalchemy.org/en/14/orm/extensions/associationproxy.html
"""
__tablename__ = "_reagenttypes_kittypes"
reagent_types_id = Column(INTEGER, ForeignKey("_reagent_types.id"), primary_key=True)
kits_id = Column(INTEGER, ForeignKey("_kits.id"), primary_key=True)
uses = Column(JSON)
required = Column(INTEGER)
last_used = Column(String(32)) #: last used lot number of this type of reagent
kit_type = relationship(KitType, back_populates="kit_reagenttype_associations")
# reference to the "ReagentType" object
reagent_type = relationship(ReagentType, back_populates="reagenttype_kit_associations")
def __init__(self, kit_type=None, reagent_type=None, uses=None, required=1):
self.kit_type = kit_type
self.reagent_type = reagent_type
self.uses = uses
self.required = required
def __repr__(self) -> str:
return f"<KitTypeReagentTypeAssociation({self.kit_type} & {self.reagent_type})>"
@validates('required')
def validate_age(self, key, value):
if not 0 <= value < 2:
raise ValueError(f'Invalid required value {value}. Must be 0 or 1.')
return value
@validates('reagenttype')
def validate_reagenttype(self, key, value):
if not isinstance(value, ReagentType):
raise ValueError(f'{value} is not a reagenttype')
return value
class Reagent(Base):
"""
@@ -247,10 +262,12 @@ class Reagent(Base):
except AttributeError:
rtype = "Unknown"
return {
"name":self.name,
"type": rtype,
"lot": self.lot,
"expiry": self.expiry.strftime("%Y-%m-%d")
}
class Discount(Base):
"""
@@ -266,6 +283,9 @@ class Discount(Base):
name = Column(String(128))
amount = Column(FLOAT(2))
def __repr__(self) -> str:
return f"<Discount({self.name})>"
class SubmissionType(Base):
"""
Abstract of types of submissions.

3
src/submissions/backend/db/models/organizations.py
View File

@@ -47,3 +47,6 @@ class Contact(Base):
phone = Column(String(32)) #: contact phone number
organization = relationship("Organization", back_populates="contacts", uselist=True, secondary=orgs_contacts) #: relationship to joined organization
def __repr__(self) -> str:
return f"<Contact({self.name})>"

36
src/submissions/backend/db/models/submissions.py
View File

@@ -13,8 +13,6 @@ from sqlalchemy.ext.associationproxy import association_proxy
import uuid
from pandas import Timestamp
from dateutil.parser import parse
import pprint
from tools import check_not_nan
logger = logging.getLogger(f"submissions.{__name__}")
@@ -348,7 +346,7 @@ class BasicSample(Base):
return value
def __repr__(self) -> str:
return f"<{self.sample_type.replace('_', ' ').title(). replace(' ', '')}({self.submitter_id})>"
return f"<{self.sample_type.replace('_', ' ').title().replace(' ', '')}({self.submitter_id})>"
def set_attribute(self, name, value):
# logger.debug(f"Setting {name} to {value}")
@@ -417,56 +415,36 @@ class WastewaterSample(BasicSample):
logger.debug(f"Validating {key}: {value}")
return value or self.submitter_id
# def __init__(self, **kwargs):
# # Had a problem getting collection date from excel as text only.
# if 'collection_date' in kwargs.keys():
# logger.debug(f"Got collection_date: {kwargs['collection_date']}. Attempting parse.")
# if isinstance(kwargs['collection_date'], str):
# logger.debug(f"collection_date is a string...")
# kwargs['collection_date'] = parse(kwargs['collection_date'])
# logger.debug(f"output is {kwargs['collection_date']}")
# # Due to the plate map being populated with RSL numbers, we have to do some shuffling.
# try:
# kwargs['rsl_number'] = kwargs['submitter_id']
# except KeyError as e:
# logger.error(f"Error using {kwargs} for submitter_id")
# try:
# check = check_not_nan(kwargs['ww_full_sample_id'])
# except KeyError:
# logger.error(f"Error using {kwargs} for ww_full_sample_id")
# check = False
# if check:
# kwargs['submitter_id'] = kwargs["ww_full_sample_id"]
# super().__init__(**kwargs)
def set_attribute(self, name:str, value):
"""
Set an attribute of this object. Extends parent.
Args:
name (str): _description_
value (_type_): _description_
name (str): name of the attribute
value (_type_): value to be set
"""
# Due to the plate map being populated with RSL numbers, we have to do some shuffling.
# logger.debug(f"Input - {name}:{value}")
match name:
case "submitter_id":
# If submitter_id already has a value, stop
if self.submitter_id != None:
return
# otherwise also set rsl_number to the same value
else:
super().set_attribute("rsl_number", value)
case "ww_full_sample_id":
# If value present, set ww_full_sample_id and make this the submitter_id
if value != None:
super().set_attribute(name, value)
name = "submitter_id"
case 'collection_date':
# If this is a string use dateutils to parse into date()
if isinstance(value, str):
logger.debug(f"collection_date {value} is a string. Attempting parse...")
value = parse(value)
case "rsl_number":
if value == None:
value = self.submitter_id
# logger.debug(f"Output - {name}:{value}")
super().set_attribute(name, value)