setup

Author: Akol125

backend/src/clients.py

@@ -1,11 +1,25 @@
-"""Initialise s3, kinesis and lambda clients"""
+"""Initialise s3, kinesis, lambda and redis clients"""
 
 from boto3 import client as boto3_client
+import os
+import logging
+import redis
 
-REGION_NAME = "eu-west-2"
+REGION_NAME = os.getenv("AWS_REGION", "eu-west-2")
 
 s3_client = boto3_client("s3", region_name=REGION_NAME)
 kinesis_client = boto3_client("kinesis", region_name=REGION_NAME)
 lambda_client = boto3_client("lambda", region_name=REGION_NAME)
 firehose_client = boto3_client("firehose", region_name=REGION_NAME)
 sqs_client = boto3_client("sqs", region_name=REGION_NAME)
+
+REDIS_HOST = os.getenv("REDIS_HOST", "")
+REDIS_PORT = int(os.getenv("REDIS_PORT", 6379))
+
+
+logging.basicConfig(level="INFO")
+logger = logging.getLogger()
+logger.setLevel("INFO")
+logger.info(f"Connecting to Redis at {REDIS_HOST}:{REDIS_PORT}")
+
+redis_client = redis.StrictRedis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)

backend/src/models/errors.py

@@ -265,3 +265,27 @@ class RecordProcessorError(Exception):
 
     def __init__(self, diagnostics_dictionary: dict):
         self.diagnostics_dictionary = diagnostics_dictionary
+
+class VaccineTypePermissionsError(Exception):
+    """
+    Raised when a supplier tries to access a vaccine type they don't have permission for.
+    """
+
+    def __init__(self, message: str):
+        super().__init__(message)
+        self.message = message
+
+    def to_operation_outcome(self) -> dict:
+        """
+        Converts the error to a FHIR-compliant OperationOutcome resource.
+        """
+        return {
+            "resourceType": "OperationOutcome",
+            "issue": [
+                {
+                    "severity": "error",
+                    "code": "forbidden",
+                    "diagnostics": self.message,
+                }
+            ]
+        }

backend/src/models/utils/permissions.py

@@ -0,0 +1,14 @@
+from clients import redis_client
+from errors import VaccineTypePermissionsError
+import json
+
+def get_supplier_permissions(supplier: str) -> list[str]:
+    permissions_data = redis_client.hget("supplier_permissions", supplier)
+    if not permissions_data:
+        return []
+    return json.loads(permissions_data)
+
+def validate_vaccine_type_permissions(vaccine_type: str, supplier: str):
+    permissions = get_supplier_permissions(supplier)
+    if not any(vaccine_type in perm for perm in permissions):
+        raise VaccineTypePermissionsError(f"{supplier} is not allowed to access {vaccine_type}")