Extract ORM improvements from other PRs (ORM PR 1)

python/lib/db/decorators/int_datetime.py

@@ -0,0 +1,27 @@
+from datetime import datetime
+
+from sqlalchemy import Integer
+from sqlalchemy.engine import Dialect
+from sqlalchemy.types import TypeDecorator
+
+
+class IntDatetime(TypeDecorator[datetime]):
+    """
+    Decorator for a database timestamp integer type.
+    In SQL, the type will appear as 'int'.
+    In Python, the type will appear as a datetime object.
+    """
+
+    impl = Integer
+
+    def process_bind_param(self, value: datetime | None, dialect: Dialect) -> int | None:
+        if value is None:
+            return None
+
+        return int(value.timestamp())
+
+    def process_result_value(self, value: int | None | None, dialect: Dialect) -> datetime | None:
+        if value is None:
+            return None
+
+        return datetime.fromtimestamp(value)

python/lib/db/decorators/true_false_bool.py

@@ -0,0 +1,33 @@
+from typing import Literal
+
+from sqlalchemy import Enum
+from sqlalchemy.engine import Dialect
+from sqlalchemy.types import TypeDecorator
+
+
+class TrueFalseBool(TypeDecorator[bool]):
+    """
+    Decorator for a database yes/no type.
+    In SQL, the type will appear as 'true' | 'false'.
+    In Python, the type will appear as a boolean.
+    """
+
+    impl = Enum('true', 'false')
+
+    def process_bind_param(self, value: bool | None, dialect: Dialect) -> Literal['true', 'false'] | None:
+        match value:
+            case True:
+                return 'true'
+            case False:
+                return 'false'
+            case None:
+                return None
+
+    def process_result_value(self, value: Literal['true', 'false'] | None, dialect: Dialect) -> bool | None:
+        match value:
+            case 'true':
+                return True
+            case 'false':
+                return False
+            case None:
+                return None

python/lib/db/decorators/y_n_bool.py

@@ -14,7 +14,7 @@ class YNBool(TypeDecorator[bool]):
 
     impl = Enum('Y', 'N')
 
-    def process_bind_param(self, value: bool | None, dialect: Dialect):
+    def process_bind_param(self, value: bool | None, dialect: Dialect) -> Literal['Y', 'N'] | None:
         match value:
             case True:
                 return 'Y'
@@ -23,7 +23,7 @@ def process_bind_param(self, value: bool | None, dialect: Dialect):
             case None:
                 return None
 
-    def process_result_value(self, value: Literal['Y', 'N'] | None, dialect: Dialect):
+    def process_result_value(self, value: Literal['Y', 'N'] | None, dialect: Dialect) -> bool | None:
         match value:
             case 'Y':
                 return True

python/lib/db/models/candidate.py

@@ -9,6 +9,7 @@
 import lib.db.models.session as db_session
 import lib.db.models.site as db_site
 from lib.db.base import Base
+from lib.db.decorators.true_false_bool import TrueFalseBool
 from lib.db.decorators.y_n_bool import YNBool
 
 
@@ -32,7 +33,7 @@ class DbCandidate(Base):
     registered_by           : Mapped[str | None]  = mapped_column('RegisteredBy')
     user_id                 : Mapped[str]         = mapped_column('UserID')
     date_registered         : Mapped[date | None] = mapped_column('Date_registered')
-    flagged_caveatemptor    : Mapped[str | None]  = mapped_column('flagged_caveatemptor')
+    flagged_caveatemptor    : Mapped[bool | None] = mapped_column('flagged_caveatemptor', TrueFalseBool)
     flagged_reason          : Mapped[int | None]  = mapped_column('flagged_reason')
     flagged_other           : Mapped[str | None]  = mapped_column('flagged_other')
     flagged_other_status    : Mapped[str | None]  = mapped_column('flagged_other_status')

python/lib/db/models/file.py

@@ -1,11 +1,12 @@
-from datetime import date
+from datetime import date, datetime
 
 from sqlalchemy import ForeignKey
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 import lib.db.models.file_parameter as db_file_parameter
 import lib.db.models.session as db_session
 from lib.db.base import Base
+from lib.db.decorators.int_datetime import IntDatetime
 
 
 class DbFile(Base):
@@ -23,7 +24,7 @@ class DbFile(Base):
     scan_type_id                   : Mapped[int | None]   = mapped_column('MriScanTypeID')
     file_type                      : Mapped[str | None]   = mapped_column('FileType')
     inserted_by_user_id            : Mapped[str]          = mapped_column('InsertedByUserID')
-    insert_time                    : Mapped[int]          = mapped_column('InsertTime')
+    insert_time                    : Mapped[datetime]     = mapped_column('InsertTime', IntDatetime)
     source_pipeline                : Mapped[str | None]   = mapped_column('SourcePipeline')
     pipeline_date                  : Mapped[date | None]  = mapped_column('PipelineDate')
     source_file_id                 : Mapped[int | None]   = mapped_column('SourceFileID')

python/lib/db/models/file_parameter.py

@@ -1,9 +1,12 @@
+from datetime import datetime
+
 from sqlalchemy import ForeignKey
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 
 import lib.db.models.file as db_file
 import lib.db.models.parameter_type as db_parameter_type
 from lib.db.base import Base
+from lib.db.decorators.int_datetime import IntDatetime
 
 
 class DbFileParameter(Base):
@@ -13,7 +16,7 @@ class DbFileParameter(Base):
     file_id     : Mapped[int]        = mapped_column('FileID', ForeignKey('files.FileID'))
     type_id     : Mapped[int]        = mapped_column('ParameterTypeID', ForeignKey('parameter_type.ParameterTypeID'))
     value       : Mapped[str | None] = mapped_column('Value')
-    insert_time : Mapped[int]        = mapped_column('InsertTime')
+    insert_time : Mapped[datetime]   = mapped_column('InsertTime', IntDatetime)
 
     file: Mapped['db_file.DbFile'] \
         = relationship('DbFile', back_populates='parameters')

python/lib/db/models/parameter_type_category.py

@@ -0,0 +1,11 @@
+from sqlalchemy.orm import Mapped, mapped_column
+
+from lib.db.base import Base
+
+
+class DbParameterTypeCategory(Base):
+    __tablename__ = 'parameter_type_category'
+
+    id   : Mapped[int]        = mapped_column('ParameterTypeCategoryID', primary_key=True)
+    name : Mapped[str | None] = mapped_column('Name')
+    type : Mapped[str | None] = mapped_column('Type')

python/lib/db/models/parameter_type_category_rel.py

@@ -0,0 +1,10 @@
+from sqlalchemy.orm import Mapped, mapped_column
+
+from lib.db.base import Base
+
+
+class DbParameterTypeCategoryRel(Base):
+    __tablename__ = 'parameter_type_category_rel'
+
+    parameter_type_id          : Mapped[int] = mapped_column('ParameterTypeID',         primary_key=True)
+    parameter_type_category_id : Mapped[int] = mapped_column('ParameterTypeCategoryID', primary_key=True)

python/lib/db/models/session.py

@@ -8,6 +8,7 @@
 import lib.db.models.project as db_project
 import lib.db.models.site as db_site
 from lib.db.base import Base
+from lib.db.decorators.true_false_bool import TrueFalseBool
 from lib.db.decorators.y_n_bool import YNBool
 
 
@@ -47,7 +48,7 @@ class DbSession(Base):
     mri_qc_pending           : Mapped[bool]            = mapped_column('MRIQCPending', YNBool)
     mri_qc_first_change_time : Mapped[datetime | None] = mapped_column('MRIQCFirstChangeTime')
     mri_qc_last_change_time  : Mapped[datetime | None] = mapped_column('MRIQCLastChangeTime')
-    mri_caveat               : Mapped[str]             = mapped_column('MRICaveat')
+    mri_caveat               : Mapped[bool]            = mapped_column('MRICaveat', TrueFalseBool)
     language_id              : Mapped[int | None]      = mapped_column('languageID')
 
     candidate : Mapped['db_candidate.DbCandidate'] = relationship('DbCandidate', back_populates='sessions')

python/lib/db/models/sex.py

@@ -0,0 +1,9 @@
+from sqlalchemy.orm import Mapped, mapped_column
+
+from lib.db.base import Base
+
+
+class DbSex(Base):
+    __tablename__ = 'sex'
+
+    name : Mapped[str] = mapped_column('Name', primary_key=True)

python/lib/db/queries/file.py

@@ -26,6 +26,17 @@ def try_get_file_with_unique_combination(
     ).scalar_one_or_none()
 
 
+def try_get_file_with_rel_path(db: Database, rel_path: str) -> DbFile | None:
+    """
+    Get an imaging file from the database using its relative path, or return `None` if no imaging
+    file is found.
+    """
+
+    return db.execute(select(DbFile)
+        .where(DbFile.rel_path == rel_path)
+    ).scalar_one_or_none()
+
+
 def try_get_file_with_hash(db: Database, file_hash: str) -> DbFile | None:
     """
     Get an imaging file from the database using its BLAKE2b or MD5 hash, or return `None` if no

python/lib/db/queries/file_parameter.py

@@ -28,3 +28,15 @@ def try_get_parameter_value_with_file_id_parameter_name(
         .where(DbParameterType.name == parameter_name)
         .where(DbFileParameter.file_id == file_id)
     ).scalar_one_or_none()
+
+
+def try_get_file_parameter_with_file_id_type_id(db: Database, file_id: int, type_id: int) -> DbFileParameter | None:
+    """
+    Get a file parameter from the database using its file ID and type ID, or return `None` if no
+    file parameter is found.
+    """
+
+    return db.execute(select(DbFileParameter)
+        .where(DbFileParameter.type_id == type_id)
+        .where(DbFileParameter.file_id == file_id)
+    ).scalar_one_or_none()

python/lib/db/queries/parameter_type.py

@@ -4,6 +4,7 @@
 from sqlalchemy.orm import Session as Database
 
 from lib.db.models.parameter_type import DbParameterType
+from lib.db.models.parameter_type_category import DbParameterTypeCategory
 
 
 def get_all_parameter_types(db: Database) -> Sequence[DbParameterType]:
@@ -12,3 +13,25 @@ def get_all_parameter_types(db: Database) -> Sequence[DbParameterType]:
     """
 
     return db.execute(select(DbParameterType)).scalars().all()
+
+
+def try_get_parameter_type_with_name(db: Database, name: str) -> DbParameterType | None:
+    """
+    Get a parameter type from the database using its name, or return `None` if no parameter type is
+    found.
+    """
+
+    return db.execute(select(DbParameterType)
+        .where(DbParameterType.name == name)
+    ).scalar_one_or_none()
+
+
+def get_parameter_type_category_with_name(db: Database, name: str) -> DbParameterTypeCategory:
+    """
+    Get a parameter type category from the database using its name, or raise an exception if no
+    parameter type category is found.
+    """
+
+    return db.execute(select(DbParameterTypeCategory)
+        .where(DbParameterTypeCategory.name == name)
+    ).scalar_one()

python/lib/db/queries/sex.py

@@ -0,0 +1,14 @@
+from sqlalchemy import select
+from sqlalchemy.orm import Session as Database
+
+from lib.db.models.sex import DbSex
+
+
+def try_get_sex_with_name(db: Database, name: str) -> DbSex | None:
+    """
+    Try to get a sex from the database using its name, or return `None` if no sex is found.
+    """
+
+    return db.execute(select(DbSex)
+        .where(DbSex.name == name)
+    ).scalar_one_or_none()

python/lib/get_session_info.py

@@ -259,7 +259,7 @@ def create_session(
         hardcopy_request = '-',
         mri_qc_status    = '',
         mri_qc_pending   = False,
-        mri_caveat       = 'true',
+        mri_caveat       = True,
     )
 
     env.db.add(session)

python/tests/integration/scripts/test_mass_nifti_pic.py

@@ -1,5 +1,4 @@
 import os
-import time
 from datetime import datetime
 
 from lib.db.models.file import DbFile
@@ -190,9 +189,10 @@ def test_running_on_a_text_file():
         file_type           = 'txt',
         session_id          = 564,
         output_type         = 'native',
-        insert_time         = int(datetime.now().timestamp()),
+        insert_time         = datetime.now(),
         inserted_by_user_id = 'test'
     )
+
     db.add(file)
     db.commit()
 
@@ -239,7 +239,7 @@ def test_successful_run():
     file_pic_data = try_get_parameter_value_with_file_id_parameter_name(db, 2, 'check_pic_filename')
     assert file_pic_data is None
 
-    current_time = time.time()
+    current_time = datetime.now()
 
     process = run_integration_script([
         'mass_nifti_pic.py',