|
1 | | -from oracle.oracle import OracleDB |
2 | | -import logging |
3 | | -from utils.oracle.subject_selection_query_builder import SubjectSelectionQueryBuilder |
4 | | -from utils.user_tools import UserTools |
5 | | -from classes.subject import Subject |
6 | | - |
7 | | - |
8 | | -class SubjectSelector: |
9 | | - """ |
10 | | - Provides helper methods for selecting screening subjects based on preconditions |
11 | | - required by specific scenarios. |
12 | | - """ |
13 | | - |
14 | | - @staticmethod |
15 | | - def get_subject_for_manual_cease(criteria: dict) -> str: |
16 | | - """ |
17 | | - Retrieves a subject NHS number suitable for manual cease, |
18 | | - based on dynamically provided selection criteria. |
19 | | -
|
20 | | - Args: |
21 | | - criteria (dict): Dictionary of filtering conditions to select a subject. |
22 | | -
|
23 | | - Returns: |
24 | | - str: The NHS number of the selected subject. |
25 | | -
|
26 | | - Raises: |
27 | | - ValueError: If no subject is found matching the criteria. |
28 | | - """ |
29 | | - logging.info( |
30 | | - f"[SUBJECT SELECTOR] Searching for subject using criteria: {criteria}" |
31 | | - ) |
32 | | - |
33 | | - hub_code = criteria.get("subject hub code", "BCS02") |
34 | | - user_details = UserTools.retrieve_user(f"Hub Manager at {hub_code}") |
35 | | - user = UserTools.get_user_object(user_details) |
36 | | - subject = Subject() |
37 | | - |
38 | | - query_builder = SubjectSelectionQueryBuilder() |
39 | | - query, bind_vars = query_builder.build_subject_selection_query( |
40 | | - criteria=criteria, |
41 | | - user=user, |
42 | | - subject=subject, |
43 | | - ) |
44 | | - |
45 | | - logging.debug( |
46 | | - f"[SUBJECT SELECTOR] Executing query:\n{query}\nWith bind variables: {bind_vars}" |
47 | | - ) |
48 | | - result_df = OracleDB().execute_query(query, bind_vars) |
49 | | - |
50 | | - if result_df.empty: |
51 | | - raise ValueError("No subject found for manual cease.") |
52 | | - |
53 | | - nhs_number = result_df["subject_nhs_number"].iloc[0] |
54 | | - logging.info(f"[SUBJECT SELECTOR] Found subject NHS number: {nhs_number}") |
55 | | - return nhs_number |
| 1 | +from oracle.oracle import OracleDB |
| 2 | +import logging |
| 3 | +from utils.oracle.subject_selection_query_builder import SubjectSelectionQueryBuilder |
| 4 | +from utils.user_tools import UserTools |
| 5 | +from classes.subject import Subject |
| 6 | + |
| 7 | + |
| 8 | +class SubjectSelector: |
| 9 | + """ |
| 10 | + Provides helper methods for selecting screening subjects based on preconditions |
| 11 | + required by specific scenarios. |
| 12 | + """ |
| 13 | + |
| 14 | + @staticmethod |
| 15 | + def get_subject_for_manual_cease(criteria: dict) -> str: |
| 16 | + """ |
| 17 | + Retrieves a subject NHS number suitable for manual cease, |
| 18 | + based on dynamically provided selection criteria. |
| 19 | +
|
| 20 | + Args: |
| 21 | + criteria (dict): Dictionary of filtering conditions to select a subject. |
| 22 | +
|
| 23 | + Returns: |
| 24 | + str: The NHS number of the selected subject. |
| 25 | +
|
| 26 | + Raises: |
| 27 | + ValueError: If no subject is found matching the criteria. |
| 28 | + """ |
| 29 | + logging.info( |
| 30 | + f"[SUBJECT SELECTOR] Searching for subject using criteria: {criteria}" |
| 31 | + ) |
| 32 | + |
| 33 | + hub_code = criteria.get("subject hub code", "BCS02") |
| 34 | + user_details = UserTools.retrieve_user(f"Hub Manager at {hub_code}") |
| 35 | + user = UserTools.get_user_object(user_details) |
| 36 | + subject = Subject() |
| 37 | + |
| 38 | + query_builder = SubjectSelectionQueryBuilder() |
| 39 | + query, bind_vars = query_builder.build_subject_selection_query( |
| 40 | + criteria=criteria, |
| 41 | + user=user, |
| 42 | + subject=subject, |
| 43 | + ) |
| 44 | + |
| 45 | + logging.debug( |
| 46 | + f"[SUBJECT SELECTOR] Executing query:\n{query}\nWith bind variables: {bind_vars}" |
| 47 | + ) |
| 48 | + result_df = OracleDB().execute_query(query, bind_vars) |
| 49 | + |
| 50 | + if result_df.empty: |
| 51 | + raise ValueError("No subject found for manual cease.") |
| 52 | + |
| 53 | + nhs_number = result_df["subject_nhs_number"].iloc[0] |
| 54 | + logging.info(f"[SUBJECT SELECTOR] Found subject NHS number: {nhs_number}") |
| 55 | + return nhs_number |
0 commit comments