Factorize get subject session

pyproject.toml

@@ -18,6 +18,11 @@ include = [
     "python/lib/db",
     "python/lib/exception",
     "python/lib/config_file.py",
+    "python/lib/env.py",
+    "python/lib/file_system.py",
+    "python/lib/get_subject_session.py",
+    "python/lib/logging.py",
+    "python/lib/make_env.py",
     "python/lib/validate_subject_info.py",
 ]
 typeCheckingMode = "strict"

python/lib/database_lib/session_db.py

@@ -1,9 +1,11 @@
 """This class performs session table related database queries and common checks"""
 
+from typing_extensions import deprecated
 
 __license__ = "GPLv3"
 
 
+@deprecated('Use `lib.db.model.session.DbSession` instead')
 class SessionDB:
     """
     This class performs database queries for session table.
@@ -35,6 +37,7 @@ def __init__(self, db, verbose):
         self.db = db
         self.verbose = verbose
 
+    @deprecated('Use `lib.db.query.try_get_candidate_with_cand_id_visit_label` instead')
     def create_session_dict(self, cand_id, visit_label):
         """
         Queries the session table for a particular candidate ID and visit label and returns a dictionary
@@ -56,6 +59,7 @@ def create_session_dict(self, cand_id, visit_label):
 
         return results[0] if results else None
 
+    @deprecated('Use `lib.db.query.site.try_get_site_with_psc_id_visit_label` instead')
     def get_session_center_info(self, pscid, visit_label):
         """
         Get site information for a given visit.
@@ -77,6 +81,7 @@ def get_session_center_info(self, pscid, visit_label):
 
         return results[0] if results else None
 
+    @deprecated('Use `lib.get_subject_session.get_candidate_next_visit_number` instead')
     def determine_next_session_site_id_and_visit_number(self, cand_id):
         """
         Determines the next session site and visit number based on the last session inserted for a given candidate.
@@ -99,6 +104,7 @@ def determine_next_session_site_id_and_visit_number(self, cand_id):
 
         return results[0] if results else None
 
+    @deprecated('Use `lib.db.model.session.DbSession` instead')
     def insert_into_session(self, fields, values):
         """
         Insert a new row in the session table using fields list as column names and values as values.

python/lib/db/connect.py

@@ -1,13 +1,11 @@
 from sqlalchemy import URL, create_engine
-from sqlalchemy.orm import Session
 
 from lib.config_file import DatabaseConfig
 
 
-def connect_to_database(config: DatabaseConfig):
+def get_database_engine(config: DatabaseConfig):
     """
-    Connect to the database and get an SQLAlchemy session to interract with it using the provided
-    credentials.
+    Connect to the database and return an SQLAlchemy engine using the provided credentials.
     """
 
     # The SQLAlchemy URL object notably escapes special characters in the configuration attributes
@@ -20,5 +18,4 @@ def connect_to_database(config: DatabaseConfig):
         database   = config.database,
     )
 
-    engine = create_engine(url)
-    return Session(engine)
+    return create_engine(url)

python/lib/db/model/notification_spool.py

@@ -25,5 +25,4 @@ class DbNotificationSpool(Base):
     origin       : Mapped[Optional[str]]      = mapped_column('Origin')
     active       : Mapped[bool]               = mapped_column('Active', YNBool)
 
-    type : Mapped['db_notification_type.DbNotificationType'] \
-        = relationship('DbNotificationType')
+    type : Mapped['db_notification_type.DbNotificationType'] = relationship('DbNotificationType')