diff --git a/enterprise_catalog/apps/api_client/algolia.py b/enterprise_catalog/apps/api_client/algolia.py
index 68517ee46..6387223b6 100644
--- a/enterprise_catalog/apps/api_client/algolia.py
+++ b/enterprise_catalog/apps/api_client/algolia.py
@@ -116,3 +116,48 @@ def replace_all_objects(self, algolia_objects):  # pragma: no cover
                 self.ALGOLIA_INDEX_NAME,
             )
             raise exc
+
+    def get_all_objects_associated_with_aggregation_key(self, aggregation_key):
+        """
+        Returns an array of Algolia object IDs associated with the given aggregation key.
+        """
+        objects = []
+        if not self.index_exists():
+            # index must exist to continue, nothing left to do
+            return objects
+        try:
+            index_browse_iterator = self.algolia_index.browse_objects({
+                "attributesToRetrieve": ["objectID"],
+                "filters": f"aggregation_key:'{aggregation_key}'",
+            })
+            for hit in index_browse_iterator:
+                objects.append(hit['objectID'])
+        except AlgoliaException as exc:
+            logger.exception(
+                'Could not retrieve objects associated with aggregation key %s due to an exception.',
+                aggregation_key,
+            )
+            raise exc
+        return objects
+
+    def remove_objects(self, object_ids):
+        """
+        Removes objects from the Algolia index.
+        """
+        if not self.index_exists():
+            # index must exist to continue, nothing left to do
+            return
+
+        try:
+            self.algolia_index.delete_objects(object_ids)
+            logger.info(
+                'The following objects were successfully removed from the %s Algolia index: %s',
+                self.ALGOLIA_INDEX_NAME,
+                object_ids,
+            )
+        except AlgoliaException as exc:
+            logger.exception(
+                'Could not remove objects from the %s Algolia index due to an exception.',
+                self.ALGOLIA_INDEX_NAME,
+            )
+            raise exc
diff --git a/enterprise_catalog/apps/catalog/algolia_content_indexing.py b/enterprise_catalog/apps/catalog/algolia_content_indexing.py
new file mode 100644
index 000000000..cf1bfef0f
--- /dev/null
+++ b/enterprise_catalog/apps/catalog/algolia_content_indexing.py
@@ -0,0 +1,26 @@
+from enterprise_catalog.apps.catalog.filters import does_query_match_content
+from enterprise_catalog.apps.catalog.models import (
+    ContentMetadata,
+    EnterpriseCatalog,
+)
+
+
+def get_catalogs_for_content(course_key, catalog_filter=None):
+    """
+    Retrieve all catalogs that contain the given course_key.
+
+    Arguments:
+        course_key (str): Course key to lookup the content metadata object for comparison against the catalog query
+            filters
+
+        catalog_filter (dict): Optional filter to apply to the catalogs queryset
+    """
+    content_object = ContentMetadata.objects.get(content_key=course_key)
+    if catalog_filter is None:
+        catalog_filter = {}
+    included_catalogs = []
+    for catalog in EnterpriseCatalog.objects.filter(**catalog_filter):
+        if does_query_match_content(catalog.catalog_query.content_filter, content_object.json_metadata):
+            included_catalogs.append(catalog)
+
+    return included_catalogs
diff --git a/enterprise_catalog/apps/catalog/tests/test_algolia_content_indexing.py b/enterprise_catalog/apps/catalog/tests/test_algolia_content_indexing.py
new file mode 100644
index 000000000..cf82f5dd1
--- /dev/null
+++ b/enterprise_catalog/apps/catalog/tests/test_algolia_content_indexing.py
@@ -0,0 +1,44 @@
+""" Tests for catalog query filtering. """
+import logging
+
+import ddt
+from django.test import TestCase
+
+from enterprise_catalog.apps.catalog.algolia_content_indexing import (
+    get_catalogs_for_content,
+)
+from enterprise_catalog.apps.catalog.tests.factories import (
+    CatalogQueryFactory,
+    ContentMetadataFactory,
+    EnterpriseCatalogFactory,
+)
+
+
+logger = logging.getLogger(__name__)
+
+
+@ddt.ddt
+class AlgoliaContentIndexingTests(TestCase):
+    """ Tests for catalog query filtering. """
+
+    def test_does_content_belong_in_catalog(self):
+        """ Test that `get_catalogs_for_content_metadata` matches content to correct catalog. """
+        content_metadata = ContentMetadataFactory(content_type='course')
+        content_key = content_metadata.json_metadata.get('key')
+        status = content_metadata.json_metadata.get('status')
+
+        included_query = CatalogQueryFactory(
+            content_filter={'content_type': 'course', 'status': status, 'key': content_key}
+        )
+        excluded_query = CatalogQueryFactory(
+            content_filter={'content_type': 'course', 'status': status, 'key__exclude': [content_key]}
+        )
+        included_catalog = EnterpriseCatalogFactory(
+            catalog_query=included_query
+        )
+        EnterpriseCatalogFactory(
+            catalog_query=excluded_query
+        )
+        found_catalogs = get_catalogs_for_content(content_key)
+        assert len(found_catalogs) == 1
+        assert found_catalogs[0] == included_catalog