refactor: moving view logic to api

openedx/core/djangoapps/content/search/api.py

@@ -6,10 +6,12 @@
 import logging
 import time
 from contextlib import contextmanager
+from datetime import datetime, timedelta, timezone
 from typing import Callable, Generator
 
 import meilisearch
 from django.conf import settings
+from django.contrib.auth import get_user_model
 from django.core.cache import cache
 from meilisearch.errors import MeilisearchError
 from meilisearch.models.task import TaskInfo
@@ -30,10 +32,13 @@
 
 log = logging.getLogger(__name__)
 
+User = get_user_model()
+
 STUDIO_INDEX_NAME = "studio_content"
 INDEX_NAME = settings.MEILISEARCH_INDEX_PREFIX + STUDIO_INDEX_NAME
 
 _MEILI_CLIENT = None
+_MEILI_API_KEY_UID = None
 
 LOCK_EXPIRE = 5 * 60  # Lock expires in 5 minutes
 
@@ -82,8 +87,8 @@ def _get_meilisearch_client():
     """
     Get the Meiliesearch client
     """
-
     global _MEILI_CLIENT  # pylint: disable=global-statement
+
     if _MEILI_CLIENT is not None:
         return _MEILI_CLIENT
 
@@ -100,6 +105,18 @@ def _get_meilisearch_client():
     return _MEILI_CLIENT
 
 
+def _get_meili_api_key_uid():
+    """
+    Helper method to get the UID of the API key we're using for Meilisearch
+    """
+    global _MEILI_API_KEY_UID  # pylint: disable=global-statement
+
+    if _MEILI_API_KEY_UID is not None:
+        return _MEILI_API_KEY_UID
+
+    _MEILI_API_KEY_UID =  _get_meilisearch_client().get_key(settings.MEILISEARCH_API_KEY).uid
+
+
 def _wait_for_meili_task(info: TaskInfo) -> None:
     """
     Simple helper method to wait for a Meilisearch task to complete
@@ -326,3 +343,29 @@ def delete_xblock_index_doc(usage_key: UsageKey) -> None:
     client = _get_meilisearch_client()
 
     _wait_for_meili_task(client.index(INDEX_NAME).delete_document(meili_id_from_opaque_key(usage_key)))
+
+
+def generate_user_token(user):
+    """
+    Returns a Meilisearch API key that only allows the user to search content that they have permission to view
+    """
+    expires_at = datetime.now(tz=timezone.utc) + timedelta(days=7)
+    search_rules = {
+        INDEX_NAME: {
+            # TODO: Apply filters here based on the user's permissions, so they can only search for content
+            # that they have permission to view. Example:
+            # 'filter': 'org = BradenX'
+        }
+    }
+    # Note: the following is just generating a JWT. It doesn't actually make an API call to Meilisearch.
+    restricted_api_key = _get_meilisearch_client.generate_tenant_token(
+        api_key_uid=_get_meili_api_key_uid(),
+        search_rules=search_rules,
+        expires_at=expires_at,
+    )
+
+    return {
+        "url": settings.MEILISEARCH_PUBLIC_URL,
+        "index_name": INDEX_NAME,
+        "api_key": restricted_api_key,
+    }

openedx/core/djangoapps/content/search/views.py

@@ -1,34 +1,23 @@
 """
 REST API for content search
 """
-from datetime import datetime, timedelta, timezone
 import logging
 
 from django.conf import settings
 from django.contrib.auth import get_user_model
-import meilisearch
 from rest_framework.exceptions import NotFound, PermissionDenied
 from rest_framework.response import Response
 from rest_framework.views import APIView
 
 from common.djangoapps.student.roles import GlobalStaff
 from openedx.core.lib.api.view_utils import view_auth_classes
-from openedx.core.djangoapps.content.search.documents import STUDIO_INDEX_NAME
+
+from . import api
 
 User = get_user_model()
 log = logging.getLogger(__name__)
 
 
-def _get_meili_api_key_uid():
-    """
-    Helper method to get the UID of the API key we're using for Meilisearch
-    """
-    if not hasattr(_get_meili_api_key_uid, "uid"):
-        client = meilisearch.Client(settings.MEILISEARCH_URL, settings.MEILISEARCH_API_KEY)
-        _get_meili_api_key_uid.uid = client.get_key(settings.MEILISEARCH_API_KEY).uid
-    return _get_meili_api_key_uid.uid
-
-
 @view_auth_classes(is_authenticated=True)
 class StudioSearchView(APIView):
     """
@@ -45,27 +34,7 @@ def get(self, request):
             # Until we enforce permissions properly (see below), this endpoint is restricted to global staff,
             # because it lets you search data from any course/library.
             raise PermissionDenied("For the moment, use of this search preview is restricted to global staff.")
-        client = meilisearch.Client(settings.MEILISEARCH_URL, settings.MEILISEARCH_API_KEY)
-        index_name = settings.MEILISEARCH_INDEX_PREFIX + STUDIO_INDEX_NAME
 
-        # Create an API key that only allows the user to search content that they have permission to view:
-        expires_at = datetime.now(tz=timezone.utc) + timedelta(days=7)
-        search_rules = {
-            index_name: {
-                # TODO: Apply filters here based on the user's permissions, so they can only search for content
-                # that they have permission to view. Example:
-                # 'filter': 'org = BradenX'
-            }
-        }
-        # Note: the following is just generating a JWT. It doesn't actually make an API call to Meilisearch.
-        restricted_api_key = client.generate_tenant_token(
-            api_key_uid=_get_meili_api_key_uid(),
-            search_rules=search_rules,
-            expires_at=expires_at,
-        )
+        response_data = api.generate_user_token(request.user)
 
-        return Response({
-            "url": settings.MEILISEARCH_PUBLIC_URL,
-            "index_name": index_name,
-            "api_key": restricted_api_key,
-        })
+        return Response(response_data)