feat: add SearchAfterMixin for ES search_after capability

course_discovery/apps/course_metadata/models.py

@@ -1149,6 +1149,62 @@ def search(cls, query, queryset=None):
         return filtered_queryset
 
 
+class SearchAfterMixin:
+    """
+    Represents objects that have a search method to query Elasticsearch and load by primary key.
+    Supports paginated results using `search_after`.
+    """
+
+    @classmethod
+    def search(cls, query, queryset=None, search_after=None, size=10000, sort_fields=None):
+        """ Queries the search index with optional pagination using `search_after`.
+
+        Args:
+            query (str): Elasticsearch querystring (e.g., `title:intro*`).
+            queryset (models.QuerySet): Base queryset to filter, defaults to objects.all().
+            search_after (list): Pointer to the last sort value of the previous page.
+            size (int): Number of results per page. 10k is the default MAX_RESULT_WINDOW size.
+            sort_fields (list): Fields to sort by, defaults to ['id'].
+
+        Returns:
+            tuple: Filtered QuerySet and `search_after` values for the next page.
+        """
+        query = clean_query(query)
+
+        if queryset is None:
+            queryset = cls.objects.all()
+
+        if query == '(*)':
+            # Early-exit optimization. Wildcard searching is very expensive in Elasticsearch. And since we just
+            # want everything, we don't need to actually query Elasticsearch at all.
+            return queryset, None
+
+        logger.info(f"Attempting Elasticsearch document search against query: {query}")
+        es_document, *_ = registry.get_documents(models=(cls,))
+        dsl_query = ESDSLQ('query_string', query=query, analyze_wildcard=True)
+
+        if sort_fields is None:
+            sort_fields = ['id']
+        search = es_document.search().query(dsl_query).sort(*sort_fields).extra(size=size)
+
+        if search_after:
+            search = search.extra(search_after=search_after)
+
+        try:
+            results = search.execute()
+        except RequestError as exp:
+            logger.warning('Elasticsearch request failed. Got exception: %r', exp)
+            results = []
+
+        ids = {result.pk for result in results}
+        next_search_after = results[-1].meta.sort if results else None
+
+        logger.info(f'{len(ids)} records extracted from Elasticsearch query "{query}"')
+        filtered_queryset = queryset.filter(pk__in=ids)
+        logger.info(f'Filtered queryset of length {len(filtered_queryset)} extracted against query "{query}"')
+        return filtered_queryset, next_search_after
+
+
 class Collaborator(TimeStampedModel):
     """
     Collaborator model, defining any collaborators who helped write course content.