CDD-3099 update non public caching

sonar-project.properties → .sonarcloud.properties

@@ -9,3 +9,6 @@ sonar.organization=ukhsa-internal
 
 # Exclude other generated/auto-generated files
 sonar.exclusions=**/migrations/**,**/__pycache__/**
+
+# Exclude files from duplication check
+sonar.cpd.exclusions=tests/unit/public_api/views/test_base.py,tests/unit/public_api/v2/views/test_base.py

caching/private_api/decorators.py

@@ -5,6 +5,7 @@
 from rest_framework.response import Response
 
 from caching.private_api.management import CacheManagement, CacheMissError
+from common.auth.cognito_jwt import backend
 
 
 class CacheCheckResultedInMissError(Exception): ...
@@ -14,7 +15,11 @@ def is_caching_v2_enabled() -> bool:
     return os.environ.get("CACHING_V2_ENABLED", "").lower() in {"true", "1"}
 
 
-def cache_response(*, timeout: int | None = None, is_reserved_namespace: bool = False):
+def cache_response(
+    *,
+    timeout: int | None = None,
+    is_reserved_namespace: bool = False,
+):
     """Decorator to wrap API views to use a previously cached response. Otherwise, calculate and save on the way out.
 
     Notes:
@@ -49,17 +54,33 @@ def cache_response(*, timeout: int | None = None, is_reserved_namespace: bool =
     def decorator(view_function):
         @wraps(view_function)
         def wrapped_view(*args, **kwargs) -> Response:
+
+            request = args[1]
+            is_public = not (_check_if_valid_non_public_request(request=request))
+
             return _retrieve_response_from_cache_or_calculate(
-                view_function, timeout, is_reserved_namespace, *args, **kwargs
+                view_function,
+                timeout,
+                is_reserved_namespace,
+                is_public,
+                *args,
+                **kwargs,
             )
 
         return wrapped_view
 
     return decorator
 
 
+def _check_if_valid_non_public_request(request) -> bool:
+    auth = backend.JSONWebTokenAuthentication()
+    result = auth.authenticate(request)
+
+    return result is not None
+
+
 def _retrieve_response_from_cache_or_calculate(
-    view_function, timeout, is_reserved_namespace, *args, **kwargs
+    view_function, timeout, is_reserved_namespace, is_public, *args, **kwargs
 ) -> Response:
     """Gets the response from the cache, otherwise recalculates from the view
 
@@ -68,6 +89,10 @@ def _retrieve_response_from_cache_or_calculate(
         then the response will always be recalculated from the server
         and no caching will take place.
 
+        If data is not public (i.e. is_public is set to "false"), then the
+        response will always be recalculated from the server and no caching
+        will take place.
+
     Args:
         view_function: The view associated with the endpoint
         timeout: The number of seconds after which the response is expired
@@ -83,9 +108,15 @@ def _retrieve_response_from_cache_or_calculate(
 
     """
     request: Request = args[1]
+    if not is_public:
+        return _calculate_response_from_view(
+            view_function, *args, is_public=is_public, **kwargs
+        )
 
     if is_caching_v2_enabled() and not is_reserved_namespace:
-        return _calculate_response_from_view(view_function, *args, **kwargs)
+        return _calculate_response_from_view(
+            view_function, *args, is_public=is_public, **kwargs
+        )
 
     cache_management = kwargs.pop(
         "cache_management",
@@ -114,7 +145,9 @@ def _retrieve_response_from_cache_or_calculate(
 def _calculate_response_and_save_in_cache(
     view_function, timeout, cache_management, cache_entry_key, *args, **kwargs
 ) -> Response:
-    response: Response = _calculate_response_from_view(view_function, *args, **kwargs)
+    response: Response = _calculate_response_from_view(
+        view_function, *args, is_public=True, **kwargs
+    )
     if timeout == 0:
         return response
 
@@ -124,5 +157,10 @@ def _calculate_response_and_save_in_cache(
     return response
 
 
-def _calculate_response_from_view(view_function, *args, **kwargs) -> Response:
-    return view_function(*args, **kwargs)
+def _calculate_response_from_view(
+    view_function, *args, is_public: bool = True, **kwargs
+) -> Response:
+    response = view_function(*args, **kwargs)
+    if not (is_public):
+        response["Cache-Control"] = "private, no-cache"
+    return response

public_api/version_02/views/base.py

@@ -3,6 +3,7 @@
 from rest_framework.request import Request
 from rest_framework.response import Response
 
+from common.auth.cognito_jwt import backend
 from public_api.metrics_interface.interface import MetricsPublicAPIInterface
 from public_api.version_02.serializers.api_time_series_request_serializer import (
     APITimeSeriesDTO,
@@ -39,4 +40,11 @@ def get(self, request: Request, *args, **kwargs) -> Response:
         )
 
         serializer = self.get_serializer(timeseries_dto_slice, many=True)
-        return Response(serializer.data)
+        response = Response(data=serializer.data)
+
+        auth = backend.JSONWebTokenAuthentication()
+        is_valid_non_public_request = auth.authenticate(request)
+        if is_valid_non_public_request:
+            response["Cache-Control"] = "private, no-cache"
+
+        return response

public_api/views/base.py

@@ -3,6 +3,7 @@
 from rest_framework.request import Request
 from rest_framework.response import Response
 
+from common.auth.cognito_jwt import backend
 from public_api.metrics_interface.interface import MetricsPublicAPIInterface
 from public_api.serializers.api_time_series_request_serializer import (
     APITimeSeriesDTO,
@@ -31,6 +32,7 @@ def _build_request_serializer(
 
     @extend_schema(tags=[PUBLIC_API_TAG])
     def get(self, request: Request, *args, **kwargs) -> Response:
+
         serializer: APITimeSeriesRequestSerializer = self._build_request_serializer(
             request=request
         )
@@ -39,4 +41,11 @@ def get(self, request: Request, *args, **kwargs) -> Response:
         )
 
         serializer = self.get_serializer(timeseries_dto_slice, many=True)
-        return Response(serializer.data)
+        response = Response(data=serializer.data)
+
+        auth = backend.JSONWebTokenAuthentication()
+        is_valid_non_public_request = auth.authenticate(request)
+        if is_valid_non_public_request:
+            response["Cache-Control"] = "private, no-cache"
+
+        return response