Moments before disaster.

CHANGELOG.md

@@ -1,3 +1,7 @@
+## 202404.05
+
+- Addition of default query method using Kwargs.
+
 ## 202404.04
 
 - Storing of default values in db rather than hardcoded.

TODO.md

@@ -1,5 +1,7 @@
+- [ ] Create a default info return function.
+- [ ] Parse comment from excel sheet.
 - [ ] Make reporting better. 
-- [ ] Build master query method?
+- [x] Build master query method?
   - Obviously there will need to be extensions, but I feel the attr method I have in Submissions could work.
 - [x] Fix Artic RSLNamer
 - [x] Put "Not applicable" reagents in to_dict() method.

src/submissions/backend/db/models/__init__.py

@@ -1,26 +1,32 @@
 '''
 Contains all models for sqlalchemy
 '''
-import sys
+from __future__ import annotations
+import sys, logging
 from sqlalchemy.orm import DeclarativeMeta, declarative_base, Query, Session
 from sqlalchemy.ext.declarative import declared_attr
+from sqlalchemy.exc import ArgumentError
 from typing import Any, List
 from pathlib import Path
+
 # Load testing environment
 if 'pytest' in sys.modules:
     from pathlib import Path
+
     sys.path.append(Path(__file__).parents[4].absolute().joinpath("tests").__str__())
 
 Base: DeclarativeMeta = declarative_base()
 
+logger = logging.getLogger(f"submissions.{__name__}")
+
 
 class BaseClass(Base):
     """
     Abstract class to pass ctx values to all SQLAlchemy objects.
     """
     __abstract__ = True  #: Will not be added to DB
-    
-    __table_args__ = {'extend_existing': True}   #: Will only add new columns
+
+    __table_args__ = {'extend_existing': True}  #: Will only add new columns
 
     @declared_attr
     def __tablename__(cls) -> str:
@@ -29,7 +35,7 @@ class BaseClass(Base):
 
         Returns:
             str: lower case class name
-        """        
+        """
         return f"_{cls.__name__.lower()}"
 
     @declared_attr
@@ -39,7 +45,7 @@ class BaseClass(Base):
 
         Returns:
             Session: DB session from ctx settings.
-        """        
+        """
         if not 'pytest' in sys.modules:
             from tools import ctx
         else:
@@ -53,13 +59,13 @@ class BaseClass(Base):
 
         Returns:
             Path: Location of the Submissions directory in Settings object
-        """        
+        """
         if not 'pytest' in sys.modules:
             from tools import ctx
         else:
             from test_settings import ctx
         return ctx.directory_path
-    
+
     @declared_attr
     def __backup_path__(cls) -> Path:
         """
@@ -67,7 +73,7 @@ class BaseClass(Base):
 
         Returns:
             Path: Location of the Submissions backup directory in Settings object
-        """        
+        """
         if not 'pytest' in sys.modules:
             from tools import ctx
         else:
@@ -75,7 +81,25 @@ class BaseClass(Base):
         return ctx.backup_path
 
     @classmethod
-    def execute_query(cls, query: Query, limit: int = 0) -> Any | List[Any]:
+    def get_default_info(cls, *args) -> dict | List[str]:
+        dicto = dict(singles=['id'])
+        output = {}
+        for k, v in dicto.items():
+            if len(args) > 0 and k not in args:
+                # logger.debug(f"Don't want {k}")
+                continue
+            else:
+                output[k] = v
+        if len(args) == 1:
+            return output[args[0]]
+        return output
+
+    @classmethod
+    def query(cls, **kwargs):
+        return cls.execute_query(**kwargs)
+
+    @classmethod
+    def execute_query(cls, query: Query = None, model=None, limit: int = 0, **kwargs) -> Any | List[Any]:
         """
         Execute sqlalchemy query.
 
@@ -85,7 +109,24 @@ class BaseClass(Base):
 
         Returns:
             Any | List[Any]: Single result if limit = 1 or List if other.
-        """        
+        """
+        if model is None:
+            model = cls
+        if query is None:
+            query: Query = cls.__database_session__.query(model)
+        # logger.debug(f"Grabbing singles using {model.get_default_info}")
+        singles = model.get_default_info('singles')
+        logger.debug(f"Querying: {model}, singles: {singles}")
+        for k, v in kwargs.items():
+            logger.debug(f"Using key: {k} with value: {v}")
+            # logger.debug(f"That key found attribute: {attr} with type: {attr}")
+            try:
+                attr = getattr(model, k)
+                query = query.filter(attr == v)
+            except (ArgumentError, AttributeError) as e:
+                logger.error(f"Attribute {k} available due to:\n\t{e}\nSkipping.")
+            if k in singles:
+                limit = 1
         with query.session.no_autoflush:
             match limit:
                 case 0:
@@ -95,6 +136,10 @@ class BaseClass(Base):
                 case _:
                     return query.limit(limit).all()
 
+    @classmethod
+    def default_info_return(cls, info, *args):
+        return info
+
     def save(self):
         """
         Add the object to the database and commit

src/submissions/backend/db/models/kits.py

@@ -947,7 +947,8 @@ class SubmissionReagentAssociation(BaseClass):
         match submission:
             case BasicSubmission() | str():
                 if isinstance(submission, str):
-                    submission = BasicSubmission.query(rsl_number=submission)
+                    # submission = BasicSubmission.query(rsl_number=submission)
+                    submission = BasicSubmission.query(rsl_plate_num=submission)
                 # logger.debug(f"Lookup SubmissionReagentAssociation by submission BasicSubmission {submission}")
                 query = query.filter(cls.submission==submission)
             case int():

src/submissions/backend/excel/parser.py

@@ -100,7 +100,7 @@ class SheetParser(object):
         Enforce that the parser has an extraction kit
         """    
         from frontend.widgets.pop_ups import ObjectSelector
-        if not check_not_nan(self.sub['extraction_kit']['value']):
+        if 'extraction_kit' not in self.sub.keys() or not check_not_nan(self.sub['extraction_kit']['value']):
             dlg = ObjectSelector(title="Kit Needed", message="At minimum a kit is needed. Please select one.", obj_type=KitType)
             if dlg.exec():
                 self.sub['extraction_kit'] = dict(value=dlg.parse_form(), missing=True)
@@ -192,13 +192,18 @@ class InfoParser(object):
             for k, v in self.map.items():
                 # exclude from generic parsing
                 if k in exclude_from_generic:
+                    logger.warning(f"Key {k} is excluded due to parser_ignore")
                     continue
                 # If the value is hardcoded put it in the dictionary directly.
                 if isinstance(v, str):
                     dicto[k] = dict(value=v, missing=False)
                     continue
                 logger.debug(f"Looking for {k} in self.map")
-                if sheet in self.map[k]['sheets']:
+                try:
+                    check = sheet in self.map[k]['sheets']
+                except TypeError:
+                    continue
+                if check:
                     relevant[k] = v
             logger.debug(f"relevant map for {sheet}: {pformat(relevant)}")
             if relevant == {}:
@@ -592,7 +597,7 @@ class PCRParser(object):
         self.plate_num = namer.parsed_name
         self.submission_type = namer.submission_type
         logger.debug(f"Set plate number to {self.plate_num} and type to {self.submission_type}")
-        parser = BasicSubmission.find_polymorphic_subclass(self.submission_type)
+        parser = BasicSubmission.find_polymorphic_subclass(polymorphic_identity=self.submission_type)
         self.samples = parser.parse_pcr(xl=self.xl, rsl_number=self.plate_num)
         
     def parse_general(self, sheet_name:str):

src/submissions/backend/validators/pydant.py

@@ -576,6 +576,7 @@ class PydSubmission(BaseModel, extra='allow'):
                             association.save()
                             logger.debug(f"Equipment association SQL object to be added to submission: {association.__dict__}")
                             instance.submission_equipment_associations.append(association)
+                # TODO: case item if item in instance.jsons()
                 case _:
                     try:
                         instance.set_attribute(key=key, value=value)

src/submissions/frontend/widgets/submission_details.py

@@ -82,7 +82,8 @@ class SubmissionDetails(QDialog):
         """        
         logger.debug(f"Details for: {submission}")
         if isinstance(submission, str):
-            submission = BasicSubmission.query(rsl_number=submission)
+            # submission = BasicSubmission.query(rsl_number=submission)
+            submission = BasicSubmission.query(rsl_plate_num=submission)
         self.base_dict = submission.to_dict(full_data=True)
         logger.debug(f"Submission details data:\n{pformat({k:v for k,v in self.base_dict.items() if k != 'samples'})}")
         # don't want id
@@ -103,7 +104,8 @@ class SubmissionDetails(QDialog):
     def sign_off(self, submission:str|BasicSubmission):
         logger.debug(f"Signing off on {submission} - ({getuser()})")
         if isinstance(submission, str):
-            submission = BasicSubmission.query(rsl_number=submission)
+            # submission = BasicSubmission.query(rsl_number=submission)
+            submission = BasicSubmission.query(rsl_plate_number=submission)
         submission.signed_by = getuser()
         submission.save()
         self.submission_details(submission=self.rsl_plate_num)

src/submissions/frontend/widgets/submission_table.py

@@ -166,9 +166,8 @@ class SubmissionsSheet(QTableView):
             for ii in range(6, len(run)):
                 new_run[f"column{str(ii-5)}_vol"] = run[ii]
             # Lookup imported submissions
-            # sub = lookup_submission_by_rsl_num(ctx=obj.ctx, rsl_num=new_run['rsl_plate_num'])
-            # sub = lookup_submissions(ctx=obj.ctx, rsl_number=new_run['rsl_plate_num'])
-            sub = BasicSubmission.query(rsl_number=new_run['rsl_plate_num'])
+            # sub = BasicSubmission.query(rsl_number=new_run['rsl_plate_num'])
+            sub = BasicSubmission.query(rsl_plate_num=new_run['rsl_plate_num'])
             # If no such submission exists, move onto the next run
             if sub == None:
                 continue