Adding more utils around scenario 1

Author: adrianoaru-nhs

classes/episode_result_type.py

@@ -1,46 +1,116 @@
+from typing import Optional, Dict
+
+
 class EpisodeResultType:
-    """
-    Utility class for mapping episode result type descriptions to logical flags or valid value IDs.
-
-    This class provides:
-        - Logical flags for "null", "not_null", and "any_surveillance_non_participation".
-        - A mapping from descriptive result labels (e.g., "normal", "abnormal", "surveillance offered") to internal valid value IDs.
-        - A method to convert descriptions to flags or IDs.
-
-    Methods:
-        from_description(description: str) -> str | int:
-            Returns the logical flag or the valid value ID for a given description.
-            Raises ValueError if the description is not recognized.
-    """
-
-    NULL = "null"
-    NOT_NULL = "not_null"
-    ANY_SURVEILLANCE_NON_PARTICIPATION = "any_surveillance_non_participation"
-
-    _label_to_id = {
-        "normal": 9501,
-        "abnormal": 9502,
-        "surveillance offered": 9503,
-        # Add real mappings as needed
-    }
+    NO_RESULT = (20311, "No Result")
+    NORMAL = (20312, "Normal (No Abnormalities Found)")
+    LOW_RISK_ADENOMA = (20314, "Low-risk Adenoma")
+    INTERMEDIATE_RISK_ADENOMA = (20315, "Intermediate-risk Adenoma")
+    HIGH_RISK_ADENOMA = (20316, "High-risk Adenoma")
+    CANCER_DETECTED = (20317, "Cancer Detected")
+    ABNORMAL = (20313, "Abnormal")
+    CANCER_NOT_CONFIRMED = (305001, "Cancer not confirmed")
+    HIGH_RISK_FINDINGS = (305606, "High-risk findings")
+    LNPCP = (305607, "LNPCP")
+    BOWEL_SCOPE_NON_PARTICIPATION = (605002, "Bowel scope non-participation")
+    FOBT_INADEQUATE_PARTICIPATION = (605003, "FOBt inadequate participation")
+    DEFINITIVE_NORMAL_FOBT_OUTCOME = (605005, "Definitive normal FOBt outcome")
+    DEFINITIVE_ABNORMAL_FOBT_OUTCOME = (605006, "Definitive abnormal FOBt outcome")
+    HIGH_RISK_FINDINGS_SURVEILLANCE_NON_PARTICIPATION = (
+        305619,
+        "High-risk findings Surveillance non-participation",
+    )
+    LNPCP_SURVEILLANCE_NON_PARTICIPATION = (
+        305618,
+        "LNPCP Surveillance non-participation",
+    )
+    HIGH_RISK_SURVEILLANCE_NON_PARTICIPATION = (
+        605004,
+        "High-risk Surveillance non-participation",
+    )
+    INTERMEDIATE_RISK_SURVEILLANCE_NON_PARTICIPATION = (
+        605007,
+        "Intermediate-risk Surveillance non-participation",
+    )
+    LYNCH_NON_PARTICIPATION = (305688, "Lynch non-participation")
+    ANY_SURVEILLANCE_NON_PARTICIPATION = (0, "(Any) Surveillance non-participation")
+    NULL = (0, "Null")
+    NOT_NULL = (0, "Not Null")
+
+    _all_types = [
+        NO_RESULT,
+        NORMAL,
+        LOW_RISK_ADENOMA,
+        INTERMEDIATE_RISK_ADENOMA,
+        HIGH_RISK_ADENOMA,
+        CANCER_DETECTED,
+        ABNORMAL,
+        CANCER_NOT_CONFIRMED,
+        HIGH_RISK_FINDINGS,
+        LNPCP,
+        BOWEL_SCOPE_NON_PARTICIPATION,
+        FOBT_INADEQUATE_PARTICIPATION,
+        DEFINITIVE_NORMAL_FOBT_OUTCOME,
+        DEFINITIVE_ABNORMAL_FOBT_OUTCOME,
+        HIGH_RISK_FINDINGS_SURVEILLANCE_NON_PARTICIPATION,
+        LNPCP_SURVEILLANCE_NON_PARTICIPATION,
+        HIGH_RISK_SURVEILLANCE_NON_PARTICIPATION,
+        INTERMEDIATE_RISK_SURVEILLANCE_NON_PARTICIPATION,
+        LYNCH_NON_PARTICIPATION,
+        ANY_SURVEILLANCE_NON_PARTICIPATION,
+        NULL,
+        NOT_NULL,
+    ]
+
+    _descriptions: Dict[str, "EpisodeResultType"] = {}
+    _lowercase_descriptions: Dict[str, "EpisodeResultType"] = {}
+    _valid_value_ids: Dict[int, "EpisodeResultType"] = {}
+
+    def __init__(self, valid_value_id: int, description: str):
+        self.valid_value_id = valid_value_id
+        self.description = description
+
+    @classmethod
+    def _init_types(cls):
+        if not cls._descriptions:
+            for valid_value_id, description in cls._all_types:
+                instance = cls(valid_value_id, description)
+                cls._descriptions[description] = instance
+                cls._lowercase_descriptions[description.lower()] = instance
+                # Only map the first occurrence of each valid_value_id
+                if valid_value_id not in cls._valid_value_ids:
+                    cls._valid_value_ids[valid_value_id] = instance
 
     @classmethod
-    def from_description(cls, description: str):
-        """
-        Returns the logical flag or the valid value ID for a given description.
-
-        Args:
-            description (str): The episode result type description.
-
-        Returns:
-            str | int: The logical flag or the valid value ID.
-
-        Raises:
-            ValueError: If the description is not recognized.
-        """
-        key = description.strip().lower()
-        if key in {cls.NULL, cls.NOT_NULL, cls.ANY_SURVEILLANCE_NON_PARTICIPATION}:
-            return key
-        if key in cls._label_to_id:
-            return cls._label_to_id[key]
-        raise ValueError(f"Unknown episode result type: '{description}'")
+    def by_description(cls, description: str) -> Optional["EpisodeResultType"]:
+        cls._init_types()
+        return cls._descriptions.get(description)
+
+    @classmethod
+    def by_description_case_insensitive(
+        cls, description: str
+    ) -> Optional["EpisodeResultType"]:
+        cls._init_types()
+        return cls._lowercase_descriptions.get(description.lower())
+
+    @classmethod
+    def by_valid_value_id(cls, valid_value_id: int) -> Optional["EpisodeResultType"]:
+        cls._init_types()
+        return cls._valid_value_ids.get(valid_value_id)
+
+    def get_id(self) -> int:
+        return self.valid_value_id
+
+    def get_description(self) -> str:
+        return self.description
+
+    def __eq__(self, other):
+        if isinstance(other, EpisodeResultType):
+            return (
+                self.valid_value_id == other.valid_value_id
+                and self.description == other.description
+            )
+        return False
+
+    def __repr__(self):
+        return f"EpisodeResultType({self.valid_value_id}, '{self.description}')"

classes/organisation_complex.py

@@ -1,33 +1,17 @@
+from typing import Optional
+
+
 class Organisation:
     """
     Class representing an organisation with id, name, and code.
     """
 
-    def __init__(self, new_id: int, new_name: str, new_code: str):
+    def __init__(
+        self,
+        new_id: Optional[int] = None,
+        new_name: Optional[str] = None,
+        new_code: Optional[str] = None,
+    ):
         self.id = new_id
         self.name = new_name
         self.code = new_code
-
-    def get_name(self) -> str:
-        """Returns the organisation name"""
-        return self.name
-
-    def set_name(self, name: str) -> None:
-        """Sets the organisation name"""
-        self.name = name
-
-    def get_id(self) -> int:
-        """Returns the organisation id"""
-        return self.id
-
-    def set_id(self, id_: int) -> None:
-        """Sets the organisation id"""
-        self.id = id_
-
-    def get_code(self) -> str:
-        """Returns the organisation code"""
-        return self.code
-
-    def set_code(self, code: str) -> None:
-        """Sets the organisation code"""
-        self.code = code

classes/repositories/database_transition_parameters.py

@@ -0,0 +1,82 @@
+from dataclasses import dataclass
+from datetime import date
+from typing import Optional
+
+
+@dataclass
+class DatabaseTransitionParameters:
+    """
+    Data class for holding parameters for a database transition.
+    """
+
+    subject_id: Optional[int] = None
+    event_id: Optional[int] = None
+    key_id: Optional[int] = None
+    appointment_inst_id: Optional[int] = None
+    kit_id: Optional[int] = None
+    transition_id: Optional[int] = None
+    screening_status_reason_id: Optional[int] = None
+    episode_status_reason_id: Optional[int] = None
+    clinical_change_reason_id: Optional[int] = None
+    notes: Optional[str] = None
+    cease_date: Optional[date] = None
+    diagnostic_test_proposed_date: Optional[date] = None
+    diagnostic_test_confirmed_date: Optional[date] = None
+    episode_closure_reason_id: Optional[int] = None
+    user_id: Optional[int] = None
+    rollback_on_failure: Optional[str] = None
+    diagnostic_test_proposed_type_id: Optional[int] = None
+    diagnostic_test_id: Optional[int] = None
+    surgery_date: Optional[date] = None
+    recall_interval: Optional[int] = None
+    discharge_id: Optional[int] = None
+    recall_calculation_method_id: Optional[int] = None
+    recall_test_id: Optional[int] = None
+    comms_initial_event_id: Optional[int] = None
+    ecomm_lett_batch_record_id: Optional[int] = None
+    ecomm_batch_letter_id: Optional[int] = None
+    intended_extend_id: Optional[int] = None
+    surgery_result_id: Optional[int] = None
+    referral_type_id: Optional[int] = None
+    additional_data_id1: Optional[int] = None
+    additional_data_date1: Optional[date] = None
+
+    def to_params_list(self) -> list:
+        """
+        Returns the parameters as a list in the order expected by the stored procedure.
+        """
+        return [
+            self.subject_id,  # p_screening_subject_id
+            self.event_id,  # p_event_id
+            self.key_id,  # p_key_id
+            self.appointment_inst_id,  # p_appointment_inst_id
+            self.kit_id,  # p_kitid
+            self.transition_id,  # p_event_transition_id
+            self.screening_status_reason_id,  # p_ss_reason_for_change_id
+            self.episode_status_reason_id,  # p_episode_status_reason_id
+            self.clinical_change_reason_id,  # p_clinical_rsn_closure_id
+            self.notes,  # p_notes
+            self.cease_date,  # p_ceased_confirm_recd_date
+            self.diagnostic_test_proposed_date,  # p_external_test_proposed_date
+            self.diagnostic_test_confirmed_date,  # p_external_test_date
+            self.episode_closure_reason_id,  # p_uncnfrmd_rsn_epis_close_id
+            self.user_id,  # p_pio_id
+            self.rollback_on_failure,  # p_rollback_on_failure
+            self.diagnostic_test_proposed_type_id,  # p_proposed_type_id
+            self.diagnostic_test_id,  # p_ext_test_id
+            self.surgery_date,  # p_surgery_date
+            self.recall_interval,  # p_alternative_scr_interval
+            self.discharge_id,  # p_pd_id
+            self.recall_calculation_method_id,  # p_recall_calculation_method_id
+            self.recall_test_id,  # p_recall_ext_test_id
+            self.comms_initial_event_id,  # p_comms_initial_event_id
+            self.ecomm_lett_batch_record_id,  # p_ecomm_lett_batch_record_id
+            self.ecomm_batch_letter_id,  # p_ecomm_batch_letter_id
+            self.intended_extend_id,  # p_intended_extent_id
+            self.surgery_result_id,  # p_surgery_result_id
+            self.referral_type_id,  # p_screening_referral_type_id
+            self.additional_data_id1,  # p_additional_data_id_1
+            self.additional_data_date1,  # p_additional_data_date_1
+            None,  # po_error_cur (OUT)
+            None,  # po_data_cur (OUT)
+        ]

classes/repositories/general_repository.py

@@ -0,0 +1,52 @@
+from classes.repositories.database_transition_parameters import (
+    DatabaseTransitionParameters,
+)
+import logging
+from utils.oracle.oracle import OracleDB
+import oracledb
+
+
+class GeneralRepository:
+    """
+    Repository for general database operations.
+    """
+
+    def __init__(self):
+        self.oracle_db = OracleDB()
+
+    def run_database_transition(
+        self, database_transition_parameters: DatabaseTransitionParameters
+    ) -> None:
+        """
+        Executes the PKG_EPISODE.p_set_episode_next_status stored procedure with the provided parameters.
+        """
+        logging.info(
+            f"Running database transition with transition_id = {database_transition_parameters.transition_id}"
+        )
+        conn = self.oracle_db.connect_to_db()
+        try:
+            cursor = conn.cursor()
+            procedure_name = "PKG_EPISODE.p_set_episode_next_status"
+            params = database_transition_parameters.to_params_list()
+            params[-2] = cursor.var(oracledb.CURSOR)  # po_error_cur
+            params[-1] = cursor.var(oracledb.CURSOR)  # po_data_cur
+
+            cursor.callproc(procedure_name, params)
+
+            # Fetch output from the cursors
+            error_cursor = params[-2].getvalue()
+            error_rows = error_cursor.fetchall() if error_cursor is not None else []
+
+            success = any(
+                "The action was performed successfully" in str(row)
+                for row in error_rows
+            )
+
+            assert success, f"Error when executing database transition: {error_rows}"
+
+            conn.commit()
+            logging.info("Database transition executed successfully.")
+        except Exception as e:
+            raise oracledb.DatabaseError(f"Error executing database transition: {e}")
+        finally:
+            self.oracle_db.disconnect_from_db(conn)

classes/repositories/subject_repository.py

@@ -82,7 +82,10 @@ def process_pi_subject(self, pio_id: int, pi_subject: PISubject) -> Optional[int
         out_params = [int, int, str]  # contact_id, error_id, error_text
 
         result = self.oracle_db.execute_stored_procedure(
-            procedure, in_params=in_params, out_params=out_params, conn=conn
+            procedure,
+            in_params=in_params,
+            out_params=out_params,
+            conn=conn,
         )
 
         new_contact_id = result.get(3)

classes/repositories/user_repository.py

@@ -0,0 +1,41 @@
+import logging
+from typing import List, Optional, Any
+from utils.oracle.oracle import OracleDB
+from classes.user_role_type import UserRoleType
+
+
+class UserRepository:
+    """
+    Repository class handling database access for users.
+    """
+
+    def __init__(self):
+        self.oracle_db = OracleDB()
+
+    def get_pio_id_for_role(self, role: "UserRoleType") -> Optional[int]:
+        """
+        Get the PIO ID for the role.
+        """
+        logging.info(f"Getting PIO ID for role: {role.user_code}")
+        sql = """
+            SELECT
+                pio.pio_id
+            FROM person_in_org pio
+            INNER JOIN person prs ON prs.prs_id = pio.prs_id
+            INNER JOIN org ON org.org_id = pio.org_id
+            WHERE prs.oe_user_code = :user_code
+              AND org.org_code = :org_code
+              AND pio.role_id = :role_id
+              AND pio.is_bcss_user = 1
+              AND TRUNC(SYSDATE) BETWEEN TRUNC(pio.start_date) AND NVL(pio.end_date, SYSDATE)
+        """
+        params = {
+            "user_code": role.user_code,
+            "org_code": role.org_code,
+            "role_id": role.role_id,
+        }
+        df = self.oracle_db.execute_query(sql, params)
+        if not df.empty:
+            pio_id = int(df["pio_id"].iloc[0])
+            return pio_id
+        return None