Migrate "get all sites from the database" to the SQLAlchemy abstraction

python/lib/database_lib/project_cohort_rel.py

@@ -1,9 +1,11 @@
 """This class performs project_cohort_rel table related database queries and common checks"""
 
+from typing_extensions import deprecated
 
 __license__ = "GPLv3"
 
 
+@deprecated('Use `lib.db.models.project_cohort.DbProjectCohort` instead')
 class ProjectCohortRel:
     """
     This class performs database queries for project_cohort_rel table.
@@ -35,6 +37,7 @@ def __init__(self, db, verbose):
         self.db = db
         self.verbose = verbose
 
+    @deprecated('Use `lib.db.models.project_cohort.DbProjectCohort` instead')
     def create_proj_cohort_rel_dict(self, project_id, cohort_id):
         """
         Get the project/cohort rel information for a given project ID and cohort ID.

python/lib/database_lib/site.py

@@ -1,9 +1,11 @@
 """This class performs database queries for the site (psc) table"""
 
+from typing_extensions import deprecated
 
 __license__ = "GPLv3"
 
 
+@deprecated('Use `lib.db.models.site.DbSite` instead')
 class Site:
     """
     This class performs database queries on the psc (site) table.
@@ -35,6 +37,7 @@ def __init__(self, db, verbose):
         self.db = db
         self.verbose = verbose
 
+    @deprecated('Use `lib.db.queries.site.get_all_sites` instead')
     def get_list_of_sites(self):
         """
         Returns a list of dictionaries storing the list of sites present in the psc table.

python/lib/db/queries/site.py

@@ -1,11 +1,14 @@
+from collections.abc import Sequence
+from typing import Optional
+
 from sqlalchemy import select
 from sqlalchemy.orm import Session as Database
 
 from lib.db.models.session import DbSession
 from lib.db.models.site import DbSite
 
 
-def try_get_site_with_cand_id_visit_label(db: Database, cand_id: int, visit_label: str):
+def try_get_site_with_cand_id_visit_label(db: Database, cand_id: int, visit_label: str) -> Optional[DbSite]:
     """
     Get a site from the database using a candidate CandID and visit label, or return `None` if no
     site is found.
@@ -16,3 +19,11 @@ def try_get_site_with_cand_id_visit_label(db: Database, cand_id: int, visit_labe
         .where(DbSession.cand_id == cand_id)
         .where(DbSession.visit_label == visit_label)
     ).scalar_one_or_none()
+
+
+def get_all_sites(db: Database) -> Sequence[DbSite]:
+    """
+    Get a sequence of all sites from the database.
+    """
+
+    return db.execute(select(DbSite)).scalars().all()

python/lib/dcm2bids_imaging_pipeline_lib/base_pipeline.py

@@ -6,14 +6,14 @@
 from lib.database import Database
 from lib.database_lib.config import Config
 from lib.db.queries.session import try_get_session_with_cand_id_visit_label
+from lib.db.queries.site import get_all_sites
 from lib.dicom_archive import DicomArchive
 from lib.exception.determine_subject_info_error import DetermineSubjectInfoError
 from lib.exception.validate_subject_info_error import ValidateSubjectInfoError
 from lib.imaging import Imaging
 from lib.imaging_upload import ImagingUpload
 from lib.logging import log_error_exit, log_verbose, log_warning
 from lib.make_env import make_env
-from lib.session import Session
 from lib.validate_subject_info import validate_subject_info
 
 
@@ -66,7 +66,6 @@ def __init__(self, loris_getopt_obj, script_name):
         self.dicom_archive_obj = DicomArchive(self.db, self.verbose)
         self.imaging_obj = Imaging(self.db, self.verbose, self.config_file)
         self.imaging_upload_obj = ImagingUpload(self.db, self.verbose)
-        self.session_obj = Session(self.db, self.verbose)
 
         # ---------------------------------------------------------------------------------------------
         # Grep config settings from the Config module
@@ -204,12 +203,12 @@ def determine_study_info(self):
 
         # if could not find center information based on cand_id and visit_label, use the
         # patient name to match it to the site alias or MRI alias
-        list_of_sites = self.session_obj.get_list_of_sites()
-        for site_dict in list_of_sites:
-            if site_dict["Alias"] in self.subject_info.name:
-                return {"CenterName": site_dict["Alias"], "CenterID": site_dict["CenterID"]}
-            elif site_dict["MRI_alias"] in self.subject_info.name:
-                return {"CenterName": site_dict["MRI_alias"], "CenterID": site_dict["CenterID"]}
+        sites = get_all_sites(self.env.db)
+        for site in sites:
+            if site.alias in self.subject_info.name:
+                return {"CenterName": site.alias, "CenterID": site.id}
+            elif site.mri_alias in self.subject_info.name:
+                return {"CenterName": site.mri_alias, "CenterID": site.id}
 
         # if we got here, it means we could not find a center associated to the dataset
         log_error_exit(

python/lib/session.py

@@ -1,9 +1,9 @@
 """This class gather functions for session handling."""
 
 from typing_extensions import deprecated
 
 from lib.database_lib.project_cohort_rel import ProjectCohortRel
 from lib.database_lib.session_db import SessionDB
 from lib.database_lib.site import Site
 
 __license__ = "GPLv3"
@@ -58,7 +58,7 @@
         """
         self.db = db
         self.verbose = verbose
 
         self.proj_cohort_rel_db_obj = ProjectCohortRel(db, verbose)
         self.session_db_obj = SessionDB(db, verbose)
         self.site_db_obj = Site(db, verbose)
@@ -194,6 +194,7 @@
         """
         return self.session_db_obj.determine_next_session_site_id_and_visit_number(cand_id)
 
+    @deprecated('Use `lib.db.queries.site.get_all_sites` instead')
     def get_list_of_sites(self):
         """
         Get the list of sites available in the psc table.
@@ -204,6 +205,7 @@
 
         return self.site_db_obj.get_list_of_sites()
 
+    @deprecated('Use `lib.db.models.project_cohort.DbProjectCohort` instead')
     def create_proj_cohort_rel_info_dict(self, project_id, cohort_id):
         """
         Populate self.proj_cohort_rel_info_dict with the content returned from the database for the ProjectID and