[PRMP-1465] Post User Restriction

lambdas/handlers/post_user_restriction_handler.py

@@ -0,0 +1,97 @@
+import json
+
+from pydantic import ValidationError
+from services.post_user_restriction_service import PostUserRestrictionService
+from utils.audit_logging_setup import LoggingService
+from utils.decorators.ensure_env_var import ensure_environment_variables
+from utils.decorators.handle_lambda_exceptions import handle_lambda_exceptions
+from utils.decorators.override_error_check import override_error_check
+from utils.decorators.set_audit_arg import set_request_context_for_logging
+from utils.exceptions import HealthcareWorkerAPIException, HealthcareWorkerPractitionerModelException, OdsErrorException, UserRestrictionException
+from utils.lambda_response import ApiGatewayResponse
+from utils.ods_utils import extract_creator_and_ods_code_from_request_context
+
+logger = LoggingService(__name__)
+
+
+def parse_body(body: str | None) -> tuple[tuple[str, str] | None, object]:
+    if not body:
+        logger.error("Missing request body")
+        return None, ApiGatewayResponse(
+            400, "Missing request body", "POST"
+        ).create_api_gateway_response()
+
+    payload = json.loads(body)
+
+    smart_card_id = payload.get("smart_card_id")
+    nhs_number = payload.get("nhs_number")
+    if not smart_card_id or not nhs_number:
+        logger.error("Missing required fields")
+        return None, ApiGatewayResponse(
+            400, "Missing required fields", "POST"
+        ).create_api_gateway_response()
+
+    return (smart_card_id, nhs_number), None
+
+
+@set_request_context_for_logging
+@override_error_check
+@ensure_environment_variables(
+    names=[
+        "RESTRICTIONS_TABLE_NAME",
+        "HEALTHCARE_WORKER_API_URL",
+    ]
+)
+@handle_lambda_exceptions
+def lambda_handler(event, context):
+    logger.info("Starting create user restriction process")
+
+    parsed, error = parse_body(event.get("body"))
+    if error:
+        return error
+    smart_card_id, nhs_number = parsed
+
+    try:
+        creator, ods_code = extract_creator_and_ods_code_from_request_context()
+    except OdsErrorException:
+        logger.error("Missing user context")
+        return ApiGatewayResponse(
+            400, "Missing user context", "POST"
+        ).create_api_gateway_response()
+
+    service = PostUserRestrictionService()
+    try:
+        response = service.create_restriction(
+            smart_card_id=smart_card_id,
+            nhs_number=nhs_number,
+            custodian=ods_code,
+            creator=creator,
+        )
+    # Translate service validation errors into a 400 response.
+    except UserRestrictionException as exc:
+        logger.error(exc)
+        return ApiGatewayResponse(
+            400, str(exc), "POST"
+        ).create_api_gateway_response()
+    except HealthcareWorkerAPIException as exc:
+        logger.error(exc)
+        return ApiGatewayResponse(
+            exc.status_code, exc.message, "POST"
+        ).create_api_gateway_response()
+    except HealthcareWorkerPractitionerModelException as exc:
+        logger.error(exc)
+        return ApiGatewayResponse(
+            400, "Unable to process restricted user information", "POST"
+        ).create_api_gateway_response()
+    except ValidationError as exc:
+        logger.error(exc)
+        return ApiGatewayResponse(
+            400, "Invalid request body", "POST"
+        ).create_api_gateway_response()
+
+    # Successful create returns the full restriction payload.
+    response_body = json.dumps(response.model_dump(by_alias=True))
+
+    return ApiGatewayResponse(
+        201, response_body, "POST"
+    ).create_api_gateway_response()

lambdas/models/user_restrictions/user_restriction_response.py

@@ -0,0 +1,34 @@
+from datetime import datetime, timezone
+
+from pydantic import BaseModel, ConfigDict, Field
+from pydantic.alias_generators import to_pascal
+
+
+class UserRestrictionResponse(BaseModel):
+    model_config = ConfigDict(
+        validate_by_alias=True,
+        validate_by_name=True,
+        alias_generator=to_pascal,
+        populate_by_name=True,
+    )
+
+    id: str = Field(alias="ID")
+    nhs_number: str = Field(alias="NHSNumber")
+    patient_name: str | None = None
+    restricted_smartcard_id: str
+    restricted_user_name: str
+    created: str
+
+    @classmethod
+    def from_restriction(cls, restriction, restricted_user_name: str) -> "UserRestrictionResponse":
+        created_iso = datetime.fromtimestamp(
+            restriction.created, tz=timezone.utc
+        ).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+        return cls(
+            id=restriction.id,
+            nhs_number=restriction.nhs_number,
+            restricted_smartcard_id=restriction.restricted_user,
+            restricted_user_name=restricted_user_name,
+            created=created_iso,
+        )

lambdas/models/user_restrictions/user_restrictions.py

@@ -33,18 +33,26 @@ class UserRestriction(BaseModel):
     creator: str = Field(
         alias=UserRestrictionsFields.CREATOR,
     )
-    removed_by: str | None = Field(
-        alias=UserRestrictionsFields.REMOVED_BY,
-        default=None,
-    )
+    removed_by: str | None = Field(alias=UserRestrictionsFields.REMOVED_BY, default=None)
     is_active: bool = Field(default=True)
     last_updated: int = Field(
         default_factory=lambda: int(datetime.now(timezone.utc).timestamp()),
     )
 
     def model_dump_camel_case(self, *args, **kwargs):
         model_dump_results = self.model_dump(*args, **kwargs)
-        camel_case_model_dump_results = {}
-        for key in model_dump_results:
-            camel_case_model_dump_results[to_camel(key)] = model_dump_results[key]
-        return camel_case_model_dump_results
+        return self.camelize(model_dump_results)
+
+    def camelize(self, model: dict) -> dict:
+        camel_case_dict = {}
+        for key, value in model.items():
+            if isinstance(value, dict):
+                return self.camelize(value)
+            if isinstance(value, list):
+                result = []
+                for item in value:
+                    result.append(self.camelize(item))
+                value = result
+            camel_case_dict[to_camel(key)] = value
+
+        return camel_case_dict

lambdas/services/post_user_restriction_service.py

@@ -0,0 +1,49 @@
+import os
+
+from models.user_restrictions.user_restriction_response import UserRestrictionResponse
+from models.user_restrictions.user_restrictions import UserRestriction, UserRestrictionsFields
+from services.base.dynamo_service import DynamoDBService
+from services.user_restrictions.healthcare_worker_service import HealthCareWorkerApiService
+from utils.audit_logging_setup import LoggingService
+from utils.exceptions import UserRestrictionException
+
+logger = LoggingService(__name__)
+
+
+class PostUserRestrictionService:
+    def __init__(self):
+        self.table_name = os.getenv("RESTRICTIONS_TABLE_NAME")
+        self.db_service = DynamoDBService()
+        self.healthcare_service = HealthCareWorkerApiService()
+
+    def create_restriction(
+        self,
+        smart_card_id: str,
+        nhs_number: str,
+        custodian: str,
+        creator: str,
+    ) -> UserRestrictionResponse:
+        if smart_card_id == creator:
+            raise UserRestrictionException("You cannot create a restriction for yourself")
+
+        practitioner = self.healthcare_service.get_practitioner(smart_card_id)
+        restricted_user_name = f"{practitioner.first_name} {practitioner.last_name}"
+
+        # Build the domain model that will be persisted to DB.
+        restriction = UserRestriction(
+            restricted_user=smart_card_id,
+            nhs_number=nhs_number,
+            custodian=custodian,
+            creator=creator,
+        )
+
+        # Persist the new restriction in the configured table.
+        self.db_service.create_item(
+            table_name=self.table_name,
+            item=restriction.model_dump(by_alias=True, exclude_none=True),
+            key_name=UserRestrictionsFields.ID.value,
+        )
+
+        logger.info("Created user restriction")
+        return UserRestrictionResponse.from_restriction(restriction, restricted_user_name)
+

lambdas/services/user_restrictions/healthcare_worker_service.py

@@ -4,6 +4,9 @@
 import requests
 from pydantic import ValidationError
 
+from models.user_restrictions.practitioner import Practitioner
+from pydantic import ValidationError
+
 from models.user_restrictions.practitioner import Practitioner
 from services.base.nhs_oauth_service import NhsOauthService
 from services.base.ssm_service import SSMService