fix: #1580 de-spaghetti-fying backend schema (#1586)

server/backend/api/app/constants.py

@@ -39,7 +39,7 @@ class AwsTargetEnv(str, Enum):
 
 # Internal defined enum client status constants for FAM 'router_forest_client'.
 # ACTIVE/INACTIVE are mapped from Forest Client API spce.
-# See schemas.py/FamForestClientStatus class.
+# See schemas/fam_forest_client_status.py class.
 class FamForestClientStatusType(str, Enum):
     ACTIVE = "A"
     INACTIVE = "I"

server/backend/api/app/crud/crud_application.py

@@ -1,7 +1,7 @@
 import logging
 from typing import List
 
-from api.app import schemas
+from api.app.schemas import RequesterSchema, FamApplicationUserRoleAssignmentGetSchema
 from api.app.constants import UserType
 from api.app.models import model as models
 from sqlalchemy import func, select
@@ -23,8 +23,8 @@ def get_application(db: Session, application_id: int):
 
 
 def get_application_role_assignments(
-    db: Session, application_id: int, requester: schemas.Requester
-) -> List[schemas.FamApplicationUserRoleAssignmentGet]:
+    db: Session, application_id: int, requester: RequesterSchema
+) -> List[FamApplicationUserRoleAssignmentGetSchema]:
     """query the user / role cross reference table to retrieve the role
     assignments.
     Delegated Admin will only see user role assignments by the roles granted for them.

server/backend/api/app/crud/crud_forest_client.py

@@ -3,7 +3,7 @@
 from api.app.models import model as models
 from sqlalchemy.orm import Session
 
-from .. import schemas
+from api.app.schemas import FamForestClientCreateSchema
 
 LOGGER = logging.getLogger(__name__)
 
@@ -22,7 +22,7 @@ def get_forest_client(db: Session, forest_client_number: str) -> models.FamFores
     return fam_forest_client
 
 
-def create_forest_client(fam_forest_client: schemas.FamForestClientCreate, db: Session):
+def create_forest_client(fam_forest_client: FamForestClientCreateSchema, db: Session):
     LOGGER.debug(f"Creating Fam_Forest_Client with: {fam_forest_client}")
 
     fam_forest_client_dict = fam_forest_client.model_dump()
@@ -45,7 +45,7 @@ def find_or_create(db: Session, forest_client_number: str, requester: str):
             "does not exist, add a new Forest Client."
         )
 
-        request_forest_client = schemas.FamForestClientCreate(
+        request_forest_client = FamForestClientCreateSchema(
             **{
                 "forest_client_number": forest_client_number,
                 "create_user": requester,

server/backend/api/app/crud/crud_role.py

@@ -1,10 +1,11 @@
 import logging
 from typing import Optional
+from sqlalchemy.orm import Session
 
 from api.app.models import model as models
+from api.app.schemas import FamForestClientCreateSchema, FamRoleCreateSchema
 from sqlalchemy.orm import Session
 
-from .. import schemas
 from . import crud_forest_client
 
 
@@ -18,7 +19,7 @@ def get_role(db: Session, role_id: int) -> Optional[models.FamRole]:
     )
 
 
-def create_role(role: schemas.FamRoleCreate, db: Session) -> models.FamRole:
+def create_role(role: FamRoleCreateSchema, db: Session) -> models.FamRole:
     LOGGER.debug(f"Creating Fam role: {role}")
 
     fam_role_dict = role.model_dump()
@@ -45,7 +46,7 @@ def create_role(role: schemas.FamRoleCreate, db: Session) -> models.FamRole:
                 + f" 327 / {forest_client_number}",
                 "create_user": fam_role_model.create_user,
             }
-            fc_pydantic = schemas.FamForestClientCreate(**fc_dict)
+            fc_pydantic = FamForestClientCreateSchema(**fc_dict)
             forest_client_model = crud_forest_client.create_forest_client(
                 db=db, fam_forest_client=fc_pydantic
             )
@@ -64,19 +65,19 @@ def create_role(role: schemas.FamRoleCreate, db: Session) -> models.FamRole:
 
 
 def get_role_by_role_name_and_app_id(
-        db: Session,
-        role_name: str,
-        application_id: int
+    db: Session, role_name: str, application_id: int
 ) -> Optional[models.FamRole]:
     """
     Gets FAM role based on role_name and application_id.
     """
-    LOGGER.debug(f"Getting FamRole by role_name: {role_name} and application_di: {application_id}")
+    LOGGER.debug(
+        f"Getting FamRole by role_name: {role_name} and application_di: {application_id}"
+    )
     return (
         db.query(models.FamRole)
         .filter(
             models.FamRole.role_name == role_name,
-            models.FamRole.application_id == application_id
+            models.FamRole.application_id == application_id,
         )
         .one_or_none()
     )

server/backend/api/app/crud/crud_user.py

@@ -7,8 +7,13 @@
 from api.app.crud import crud_utils
 from api.app.integration.idim_proxy import IdimProxyService
 from api.config import config
-
-from .. import schemas
+from api.app.schemas import (
+    FamUserSchema,
+    TargetUserSchema,
+    FamUserUpdateResponseSchema,
+    IdimProxySearchParamSchema,
+    IdimProxyBceidSearchParamSchema,
+)
 
 LOGGER = logging.getLogger(__name__)
 
@@ -74,11 +79,11 @@ def get_user_by_domain_and_guid(
     )
 
 
-def create_user(fam_user: schemas.FamUser, db: Session):
+def create_user(fam_user: FamUserSchema, db: Session):
     """used to add a new FAM user to the database
 
     :param fam_user: _description_
-    :type fam_user: schemas.FamUser
+    :type fam_user: FamUserSchema
     :param db: _description_
     :type db: Session
     :return: _description_
@@ -124,7 +129,7 @@ def find_or_create(
         # or found user by domain and username, but user guid does not match (this is the edge case that could happen when username changed from IDP provider)
         # create a new user
         else:
-            request_user = schemas.FamUser(
+            request_user = FamUserSchema(
                 **{
                     "user_type_code": user_type_code,
                     "user_name": user_name,
@@ -191,7 +196,7 @@ def update_user_name(
 def update_user_properties_from_verified_target_user(
     db: Session,
     user_id: int,
-    target_user: schemas.TargetUser,
+    target_user: TargetUserSchema,
     requester: str,  # cognito_user_id
 ):
     """
@@ -257,7 +262,7 @@ def fetch_initial_requester_info(db: Session, cognito_user_id: str):
 
 def update_user_info_from_idim_source(
     db: Session, use_pagination: bool, page: int, per_page: int
-) -> schemas.FamUserUpdateResponse:
+) -> FamUserUpdateResponseSchema:
     """
     Go through each user record in the database,
     update the user information to match the record in IDIM web service,
@@ -311,7 +316,7 @@ def update_user_info_from_idim_source(
             if user.user_type_code == UserType.IDIR:
                 # IDIM web service doesn't support search IDIR by user_guid, so we search by userID
                 search_result = idim_proxy_service.search_idir(
-                    schemas.IdimProxySearchParam(**{"userId": user.user_name})
+                    IdimProxySearchParamSchema(**{"userId": user.user_name})
                 )
 
                 if not user.user_guid:
@@ -338,7 +343,7 @@ def update_user_info_from_idim_source(
                 if user.user_guid:
                     # if found business bceid user by user_guid, update username if necessary
                     search_result = idim_proxy_service.search_business_bceid(
-                        schemas.IdimProxyBceidSearchParam(
+                        IdimProxyBceidSearchParamSchema(
                             **{
                                 "searchUserBy": IdimSearchUserParamType.USER_GUID,
                                 "searchValue": user.user_guid,
@@ -352,7 +357,7 @@ def update_user_info_from_idim_source(
                 else:
                     # if user has no user_guid in our database, find by user_name and add user_guid to database
                     search_result = idim_proxy_service.search_business_bceid(
-                        schemas.IdimProxyBceidSearchParam(
+                        IdimProxyBceidSearchParamSchema(
                             **{
                                 "searchUserBy": IdimSearchUserParamType.USER_ID,
                                 "searchValue": user.user_name,
@@ -395,7 +400,7 @@ def update_user_info_from_idim_source(
             LOGGER.debug(f"Failed to update user info: {e}")
             failed_user_list.append(user.user_id)
 
-    return schemas.FamUserUpdateResponse(
+    return FamUserUpdateResponseSchema(
         **{
             "total_db_users_count": total_db_users_count,
             "current_page": page,

server/backend/api/app/crud/crud_user_role.py

@@ -3,7 +3,14 @@
 from typing import List
 
 from api.app import constants as famConstants
-from api.app import schemas
+from api.app.schemas import (
+    FamUserRoleAssignmentCreateSchema,
+    TargetUserSchema,
+    FamUserRoleAssignmentCreateResponseSchema,
+    FamApplicationUserRoleAssignmentGetSchema,
+    GCNotifyGrantAccessEmailParamSchema,
+    FamRoleCreateSchema
+)
 from api.app.crud import crud_forest_client, crud_role, crud_user, crud_utils
 from api.app.crud.validator.forest_client_validator import (
     forest_client_active, forest_client_number_exists,
@@ -12,18 +19,17 @@
 from api.app.integration.gc_notify import GCNotifyEmailService
 from api.app.models import model as models
 from api.app.utils.utils import raise_http_exception
-from requests import HTTPError
 from sqlalchemy.orm import Session
 
 LOGGER = logging.getLogger(__name__)
 
 
 def create_user_role_assignment_many(
     db: Session,
-    request: schemas.FamUserRoleAssignmentCreate,
-    target_user: schemas.TargetUser,
+    request: FamUserRoleAssignmentCreateSchema,
+    target_user: TargetUserSchema,
     requester: str,
-) -> List[schemas.FamUserRoleAssignmentCreateResponse]:
+) -> List[FamUserRoleAssignmentCreateResponseSchema]:
     """
     Create fam_user_role_xref Association
 
@@ -74,7 +80,7 @@ def create_user_role_assignment_many(
         fam_role.role_type_code == famConstants.RoleType.ROLE_TYPE_ABSTRACT
     )
 
-    create_return_list: List[schemas.FamUserRoleAssignmentCreateResponse] = []
+    create_return_list: List[FamUserRoleAssignmentCreateResponseSchema] = []
 
     if require_child_role:
         LOGGER.debug(
@@ -155,27 +161,25 @@ def create_user_role_assignment(
     )
 
     if fam_user_role_xref:
-        error_msg = (
-            f"Role {fam_user_role_xref.role.role_name} already assigned to user {fam_user_role_xref.user.user_name}."
-        )
-        create_user_role_assginment_return = (
-            schemas.FamUserRoleAssignmentCreateResponse(
-                **{
-                    "status_code": HTTPStatus.CONFLICT,
-                    "detail": schemas.FamApplicationUserRoleAssignmentGet(**fam_user_role_xref.__dict__),
-                    "error_message": error_msg,
-                }
-            )
+        error_msg = f"Role {fam_user_role_xref.role.role_name} already assigned to user {fam_user_role_xref.user.user_name}."
+        create_user_role_assginment_return = FamUserRoleAssignmentCreateResponseSchema(
+            **{
+                "status_code": HTTPStatus.CONFLICT,
+                "detail": FamApplicationUserRoleAssignmentGetSchema(
+                    **fam_user_role_xref.__dict__
+                ),
+                "error_message": error_msg,
+            }
         )
     else:
         fam_user_role_xref = create(db, user.user_id, role.role_id, requester)
-        create_user_role_assginment_return = (
-            schemas.FamUserRoleAssignmentCreateResponse(
-                **{
-                    "status_code": HTTPStatus.OK,
-                    "detail": schemas.FamApplicationUserRoleAssignmentGet(**fam_user_role_xref.__dict__),
-                }
-            )
+        create_user_role_assginment_return = FamUserRoleAssignmentCreateResponseSchema(
+            **{
+                "status_code": HTTPStatus.OK,
+                "detail": FamApplicationUserRoleAssignmentGetSchema(
+                    **fam_user_role_xref.__dict__
+                ),
+            }
         )
 
     return create_user_role_assginment_return
@@ -267,7 +271,7 @@ def find_or_create_forest_client_child_role(
 
     if not child_role:
         child_role = crud_role.create_role(
-            schemas.FamRoleCreate(
+            FamRoleCreateSchema(
                 **{
                     "parent_role_id": parent_role.role_id,
                     "application_id": parent_role.application_id,
@@ -301,8 +305,9 @@ def find_by_id(db: Session, user_role_xref_id: int) -> models.FamUserRoleXref:
 
 
 def send_user_access_granted_email(
-        target_user: schemas.TargetUser,
-        roles_assignment_responses: List[schemas.FamUserRoleAssignmentCreateResponse]):
+    target_user: TargetUserSchema,
+    roles_assignment_responses: List[FamUserRoleAssignmentCreateResponseSchema],
+):
     """
     Send email using GC Notify integration service.
     TODO: Erro handling when sending email encountered technical errors (400/500). Ticket #1471.
@@ -325,15 +330,19 @@ def send_user_access_granted_email(
         )
         with_client_number = "yes" if roles_assignment_responses[0].detail.role.client_number is not None else "no"
         email_service = GCNotifyEmailService()
-        email_params = schemas.GCNotifyGrantAccessEmailParam(**{
-            "first_name": target_user.first_name,
-            "last_name": target_user.last_name,
-            "application_name": roles_assignment_responses[0].detail.role.application.application_description,
-            "role_list_string": granted_roles,
-            "application_team_contact_email": None,  # TODO: ticket #1507 to implement this.
-            "send_to_email": target_user.email,
-            "with_client_number": with_client_number
-        })
+        email_params = GCNotifyGrantAccessEmailParamSchema(
+            **{
+                "first_name": target_user.first_name,
+                "last_name": target_user.last_name,
+                "application_name": roles_assignment_responses[
+                    0
+                ].detail.role.application.application_description,
+                "role_list_string": granted_roles,
+                "application_team_contact_email": None,  # TODO: ticket #1507 to implement this.
+                "send_to_email": target_user.email,
+                "with_client_number": with_client_number,
+            }
+        )
 
         if granted_roles == "":  # no role is granted
             return

server/backend/api/app/crud/validator/forest_client_validator.py

@@ -2,21 +2,21 @@
 from typing import List, Union
 
 from api.app.constants import FOREST_CLIENT_STATUS
-from api.app.schemas import ForestClientIntegrationFindResponse
+from api.app.schemas import ForestClientIntegrationFindResponseSchema
 
 
 LOGGER = logging.getLogger(__name__)
 
 
 def forest_client_number_exists(
-    forest_client_find_result: List[ForestClientIntegrationFindResponse],
+    forest_client_find_result: List[ForestClientIntegrationFindResponseSchema],
 ) -> bool:
     # Exact client number search - should only contain 1 result.
     return len(forest_client_find_result) == 1
 
 
 def forest_client_active(
-    forest_client_find_result: List[ForestClientIntegrationFindResponse],
+    forest_client_find_result: List[ForestClientIntegrationFindResponseSchema],
 ) -> bool:
     return (
         (
@@ -29,7 +29,7 @@ def forest_client_active(
 
 
 def get_forest_client_status(
-    forest_client_find_result: List[ForestClientIntegrationFindResponse],
+    forest_client_find_result: List[ForestClientIntegrationFindResponseSchema],
 ) -> Union[str, None]:
     return (
         forest_client_find_result[0][FOREST_CLIENT_STATUS["KEY"]]