chore: Upgrade Python requirements

enterprise_catalog/apps/api/base/tests/enterprise_customer_views.py

@@ -64,10 +64,24 @@ def _get_generate_diff_base_url(self, enterprise_catalog_uuid=None):
         )
 
     def _get_content_metadata_base_url(self, enterprise_uuid, content_identifier):
+        """
+        Helper to construct the base url for the customer content metadata endpoint
+        """
         return reverse(
             f'api:{self.VERSION}:customer-content-metadata-retrieve',
             kwargs={
                 'enterprise_uuid': enterprise_uuid,
                 'content_identifier': content_identifier,
             },
         )
+
+    def _get_secured_algolia_api_key_base_url(self, enterprise_uuid):
+        """
+        Helper to construct the base url for the secured Algolia API key endpoint
+        """
+        return reverse(
+            f'api:{self.VERSION}:enterprise-customer-secured-algolia-api-key',
+            kwargs={
+                'enterprise_uuid': enterprise_uuid,
+            },
+        )

enterprise_catalog/apps/api/v1/serializers.py

@@ -87,6 +87,25 @@ def find_and_modify_catalog_query(
         return content_filter_from_hash
 
 
+class BaseSerializer(serializers.Serializer):
+    """
+    Base serializer.
+    """
+    def create(self, *args, **kwargs):
+        return None
+
+    def update(self, *args, **kwargs):
+        return None
+
+
+class BaseErrorSerializer(BaseSerializer):
+    """
+    Base error serializer.
+    """
+    user_message = serializers.CharField()
+    developer_message = serializers.CharField()
+
+
 class CatalogQuerySerializer(serializers.ModelSerializer):
     """
     Serializer for the `CatalogQuery` model
@@ -525,3 +544,28 @@ class Meta:
             "external_id",
             "title"
         ]
+
+
+class SecuredAlgoliaAPIKeySerializer(BaseSerializer):
+    """
+    Serializer for the secured Algolia API key and expiration.
+    """
+    secured_api_key = serializers.CharField()
+    valid_until = serializers.DateTimeField()
+
+
+class SecuredAlgoliaAPIKeyResponseSerializer(BaseSerializer):
+    """
+    Serializer for the response of the secured Algolia API key endpoint.
+    """
+    algolia = SecuredAlgoliaAPIKeySerializer(help_text='Secured Algolia API key and expiration.')
+    catalog_uuids_to_catalog_query_uuids = serializers.DictField(
+        child=serializers.UUIDField(),
+        help_text='Mapping of catalog UUIDs to catalog query UUIDs.',
+    )
+
+
+class SecuredAlgoliaAPIKeyErrorResponseSerializer(BaseErrorSerializer):
+    """
+    Serializer for the error response of the secured Algolia API key endpoint.
+    """

enterprise_catalog/apps/api/v1/tests/mixins.py

@@ -96,7 +96,7 @@ def set_up_staff(self):
         self.assign_catalog_admin_feature_role()
         self.assign_catalog_admin_jwt_role()
 
-    def set_up_catalog_learner(self):
+    def set_up_catalog_learner(self, enterprise_uuid=None):
         """
         Helper for setting up tests as a catalog learner
         """
@@ -106,7 +106,7 @@ def set_up_catalog_learner(self):
         self.role_assignment = EnterpriseCatalogRoleAssignmentFactory(
             role=self.role,
             user=self.user,
-            enterprise_id=self.enterprise_uuid
+            enterprise_id=(enterprise_uuid or self.enterprise_uuid)
         )
         self.set_jwt_cookie([(ENTERPRISE_CATALOG_LEARNER_ROLE, self.enterprise_uuid)])
 

enterprise_catalog/apps/api/v1/tests/test_enterprise_customer_views.py

@@ -3,6 +3,8 @@
 from unittest import mock
 
 import pytz
+from algoliasearch.exceptions import AlgoliaException
+from django.test import override_settings
 from rest_framework import status
 
 from enterprise_catalog.apps.api.base.tests.enterprise_customer_views import (
@@ -470,3 +472,98 @@ def test_filter_content_items_specified_catalogs(self):
 
         # response should only contain content keys found in the "second_catalog"
         self.assertEqual(response.get('filtered_content_keys'), [relevant_content_key])
+
+    @override_settings(ALGOLIA={'SEARCH_API_KEY': 'fake-search-api-search'})
+    def test_secured_algolia_api_key_generation(self):
+        """
+        Test that the secured Algolia API key is generated correctly.
+        """
+        url = self._get_secured_algolia_api_key_base_url(self.enterprise_uuid)
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        expected_catalog_to_catalog_query_mappings = {
+            str(self.enterprise_catalog.uuid): str(self.enterprise_catalog.catalog_query.uuid),
+        }
+        self.assertIsInstance(response.data['algolia']['secured_api_key'], str)
+        self.assertTrue(bool(response.data['algolia']['secured_api_key']))  # Ensures the string isn't empty
+
+        # Validate ISO format for valid_until
+        valid_until = response.data['algolia']['valid_until']
+        try:
+            datetime.fromisoformat(valid_until)  # Will raise ValueError if the format is incorrect
+        except ValueError:
+            self.fail(f"Invalid ISO format for valid_until: {valid_until}")
+
+        # Validate catalog uuids to catalog query uuids mapping
+        self.assertEqual(
+            response.data['catalog_uuids_to_catalog_query_uuids'],
+            expected_catalog_to_catalog_query_mappings,
+        )
+
+    @override_settings(ALGOLIA={})
+    def test_secured_algolia_api_key_missing_search_api_key(self):
+        """
+        Test that the secured Algolia API key is not generated if the search API key is missing.
+        """
+        url = self._get_secured_algolia_api_key_base_url(self.enterprise_uuid)
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        expected_error_response = {
+            'user_message': 'Error generating secured Algolia API key.',
+            'developer_message': (
+                'Cannot generate secured Algolia API key without the ALGOLIA.SEARCH_API_KEY in settings.'
+            ),
+        }
+        self.assertEqual(response.data, expected_error_response)
+
+    @mock.patch(
+        "algoliasearch.search_client.SearchClient.generate_secured_api_key",
+        side_effect=AlgoliaException("Mocked exception"),
+    )
+    @override_settings(ALGOLIA={'SEARCH_API_KEY': 'fake-search-api-search'})
+    def test_secured_algolia_api_key_algolia_exception(self, mock_generate_key):  # pylint: disable=unused-argument
+        """
+        Test that the secured Algolia API key is not generated if an AlgoliaException occurs.
+        """
+        url = self._get_secured_algolia_api_key_base_url(self.enterprise_uuid)
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        expected_error_response = {
+            'user_message': 'Error generating secured Algolia API key.',
+            'developer_message': 'Mocked exception',
+        }
+        self.assertEqual(response.data, expected_error_response)
+
+    def test_secured_algolia_api_key_no_catalogs(self):
+        """
+        Test that the secured Algolia API key is not generated if there are no catalogs.
+        """
+        fake_enterprise_uuid = uuid.uuid4()
+        self.set_up_catalog_learner(enterprise_uuid=fake_enterprise_uuid)
+        url = self._get_secured_algolia_api_key_base_url(enterprise_uuid=fake_enterprise_uuid)
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        expected_error_response = {
+            'user_message': 'Error generating secured Algolia API key.',
+            'developer_message': 'No enterprise catalogs found for the specified enterprise customer.',
+        }
+        self.assertEqual(response.data, expected_error_response)
+
+    def test_secured_algolia_api_key_no_catalog_queries(self):
+        """
+        Test that the secured Algolia API key is not generated if there are no catalog queries.
+        """
+        fake_enterprise_uuid = uuid.uuid4()
+        EnterpriseCatalogFactory(
+            enterprise_uuid=fake_enterprise_uuid,
+            catalog_query=None,  # Ensure no catalog query is associated with the catalog
+        )
+        self.set_up_catalog_learner(enterprise_uuid=fake_enterprise_uuid)
+        url = self._get_secured_algolia_api_key_base_url(enterprise_uuid=fake_enterprise_uuid)
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        expected_error_response = {
+            'user_message': 'Error generating secured Algolia API key.',
+            'developer_message': 'No catalog queries found for the specified enterprise customer.',
+        }
+        self.assertEqual(response.data, expected_error_response)

enterprise_catalog/apps/api/v1/views/enterprise_customer.py

@@ -1,7 +1,10 @@
 import logging
 import uuid
 
+from algoliasearch.exceptions import AlgoliaException
+from django.core.exceptions import ImproperlyConfigured
 from django.utils.decorators import method_decorator
+from drf_spectacular.utils import OpenApiExample, extend_schema
 from edx_rbac.utils import get_decoded_jwt
 from rest_framework.decorators import action
 from rest_framework.exceptions import NotFound, PermissionDenied
@@ -11,9 +14,14 @@
 from enterprise_catalog.apps.api.v1.decorators import (
     require_at_least_one_query_parameter,
 )
-from enterprise_catalog.apps.api.v1.serializers import ContentMetadataSerializer
+from enterprise_catalog.apps.api.v1.serializers import (
+    ContentMetadataSerializer,
+    SecuredAlgoliaAPIKeyErrorResponseSerializer,
+    SecuredAlgoliaAPIKeyResponseSerializer,
+)
 from enterprise_catalog.apps.api.v1.utils import unquote_course_keys
 from enterprise_catalog.apps.api.v1.views.base import BaseViewSet
+from enterprise_catalog.apps.api_client.algolia import AlgoliaSearchClient
 from enterprise_catalog.apps.catalog.models import EnterpriseCatalog
 
 
@@ -209,3 +217,122 @@ def content_metadata(self, customer_uuid, content_identifier, **kwargs):  # pyli
         """
         serializer = self.get_metadata_item_serializer()
         return Response(serializer.data)
+
+    @extend_schema(
+        summary='Get secured Algolia API key for enterprise customer',
+        description='Returns a secured Algolia API key for the given enterprise UUID. '
+                    'The key can be used within frontends to access Algolia search API, '
+                    'restricted with the enterprise\'s available catalog query UUIDs.',
+        responses={
+            200: SecuredAlgoliaAPIKeyResponseSerializer,
+            400: SecuredAlgoliaAPIKeyErrorResponseSerializer,
+        },
+        tags=['Enterprise Customer'],
+        examples=[
+            OpenApiExample(
+                'Secured Algolia API Key Response',
+                value={
+                    'algolia': {
+                        'secured_api_key': '...',
+                        'valid_until': '2025-02-28T12:00:00Z',
+                    },
+                    'catalog_uuids_to_catalog_query_uuids': {
+                        'catalog_uuid_1': 'catalog_query_uuid_1',
+                        'catalog_uuid_2': 'catalog_query_uuid_2',
+                    },
+                },
+            ),
+        ]
+    )
+    @action(detail=True, methods=['get'], url_path='secured-algolia-api-key')
+    def secured_algolia_api_key(self, request, enterprise_uuid, **kwargs):
+        """
+        Returns a secured Algolia API key for the given enterprise UUID.
+        The key can be used within frontends to access Algolia search API,
+        restricted with the enterprise's available catalog query UUIDs.
+        """
+        # Fetch all EnterpriseCatalogs associated with the given enterprise_uuid
+        enterprise_catalogs = EnterpriseCatalog.objects.filter(
+            enterprise_uuid=enterprise_uuid
+        ).select_related('catalog_query')
+        algolia_api_key_error_message = 'Error generating secured Algolia API key.'
+
+        # Return an error response if no EnterpriseCatalogs are found
+        if not enterprise_catalogs:
+            logger.error(
+                f'No enterprise catalogs found for the given enterprise UUID: {enterprise_uuid}.'
+            )
+            error_response = {
+                'user_message': algolia_api_key_error_message,
+                'developer_message': 'No enterprise catalogs found for the specified enterprise customer.',
+            }
+            return Response(
+                SecuredAlgoliaAPIKeyErrorResponseSerializer(error_response).data,
+                status=HTTP_400_BAD_REQUEST,
+            )
+
+        catalog_query_uuids = [
+            enterprise_catalog.catalog_query.uuid for enterprise_catalog in enterprise_catalogs
+            if enterprise_catalog.catalog_query
+        ]
+        # Return an error response if no CatalogQueries are found
+        if not catalog_query_uuids:
+            logger.error(
+                f'No catalog queries found for the given enterprise UUID: {enterprise_uuid}.'
+            )
+            error_response = {
+                'user_message': algolia_api_key_error_message,
+                'developer_message': 'No catalog queries found for the specified enterprise customer.'
+            }
+            return Response(
+                SecuredAlgoliaAPIKeyErrorResponseSerializer(error_response).data,
+                status=HTTP_400_BAD_REQUEST,
+            )
+
+        # Create a mapping of catalog UUIDs to their corresponding catalog query UUIDs
+        catalog_uuids_to_catalog_query_uuids = {
+            str(catalog.uuid): str(catalog.catalog_query.uuid) if catalog.catalog_query else None
+            for catalog in enterprise_catalogs
+        }
+
+        # Generate a secured Algolia API key using the AlgoliaSearchClient
+        algolia_client = AlgoliaSearchClient()
+        try:
+            result = algolia_client.generate_secured_api_key(
+                user_id=request.user.id,
+                enterprise_catalog_query_uuids=catalog_query_uuids,
+            )
+        except ImproperlyConfigured as exc:
+            logger.exception(
+                f'Secured Algolia API key generation failed for enterprise UUID {enterprise_uuid}'
+            )
+            error_response = {
+                'user_message': algolia_api_key_error_message,
+                'developer_message': exc,
+            }
+            return Response(
+                SecuredAlgoliaAPIKeyErrorResponseSerializer(error_response).data,
+                status=HTTP_400_BAD_REQUEST,
+            )
+        except AlgoliaException as exc:
+            logger.exception(
+                f'Secured Algolia API key generation failed for enterprise UUID {enterprise_uuid}'
+            )
+            error_response = {
+                'user_message': algolia_api_key_error_message,
+                'developer_message': exc
+            }
+            return Response(
+                SecuredAlgoliaAPIKeyErrorResponseSerializer(error_response).data,
+                status=HTTP_400_BAD_REQUEST,
+            )
+
+        # Serialize the response data
+        response_data = {
+            'algolia': {
+                'secured_api_key': result.get('secured_api_key'),
+                'valid_until': result.get('valid_until'),
+            },
+            'catalog_uuids_to_catalog_query_uuids': catalog_uuids_to_catalog_query_uuids,
+        }
+        return Response(SecuredAlgoliaAPIKeyResponseSerializer(response_data).data)

enterprise_catalog/apps/api/v2/views/enterprise_customer.py

@@ -1,5 +1,7 @@
 import logging
 
+from rest_framework.decorators import action
+
 from enterprise_catalog.apps.api.v1.views.enterprise_customer import (
     EnterpriseCustomerViewSet,
 )
@@ -38,3 +40,10 @@ def filter_content_keys(self, catalog, content_keys):
 
     def contains_content_keys(self, catalog, content_keys):
         return catalog.contains_content_keys(content_keys, include_restricted=True)
+
+    @action(detail=False, methods=['get'], url_path='secured-algolia-api-key')
+    def secured_algolia_api_key(self, request, enterprise_uuid, **kwargs):
+        """
+        There is no V2 version of this endpoint.
+        """
+        raise NotImplementedError('This endpoint is not available in V2.')