Adds portal.add_catalog_indexes and portal.add_catalog_metadata methods

docs/api/index.md

@@ -40,6 +40,8 @@ exceptions
     api.portal.send_email
     api.portal.show_message
     api.portal.get_registry_record
+    api.portal.add_catalog_indexes
+    api.portal.add_catalog_metadata
 
 ```
 

docs/portal.md

@@ -459,6 +459,92 @@ for vocabulary_name in common_vocabularies:
     assert vocabulary_name in vocabulary_names
 ```
 
+(portal-add-catalog-indexes-example)=
+
+## Add catalog indexes
+
+To add indexes to the portal catalog, use {meth}`api.portal.add_catalog_indexes`.
+This function returns a list of the names of the indexes that were added.
+
+The following collection of code snippets demonstrate how to add indexes and either use default logging, to skip reindexing, or to use a customer logger.
+
+```python
+from plone import api
+
+# Add a single field index
+api.portal.add_catalog_indexes([('my_custom_field', 'FieldIndex')])
+
+# Add multiple indexes with different types
+indexes_to_add = [
+    ('text_content', 'ZCTextIndex'),
+    ('tags', 'KeywordIndex')
+]
+api.portal.add_catalog_indexes(indexes_to_add)
+
+# Add indexes without reindexing
+api.portal.add_catalog_indexes([('quick_field', 'FieldIndex')], reindex=False)
+```
+
+% invisible-code-block: python
+%
+% # Verify the indexes were added to the catalog
+% catalog = api.portal.get_tool('portal_catalog')
+% self.assertIn('my_custom_field', catalog.indexes())
+% self.assertIn('tags', catalog.indexes())
+% self.assertIn('quick_field', catalog.indexes())
+
+
+### ZCTextIndex Special Handling
+
+When adding a `ZCTextIndex`, the function automatically applies additional parameters:
+
+- `lexicon_id`: Set to `plone_lexicon` by default
+- `index_type`: Set to `Okapi BM25 Rank`
+- `doc_attr`: Set to the index name provided
+
+This ensures proper configuration for text-based searching and indexing in Plone.
+
+The function returns a list of the names of the indexes that were added.
+
+(portal-add-catalog-metadata-example)=
+
+## Add catalog metadata columns
+
+To add metadata columns to the portal catalog, use {meth}`api.portal.add_catalog_metadata`.
+This function returns a list of the names of the columns that were added.
+
+```{note}
+Adding metadata columns only makes them available for storage.
+You still need to reindex your content to populate the values.
+```
+
+The following collection of code snippets adds metadata columns with either the default logger or a custom logger.
+
+```python
+from plone import api
+
+# Add new metadata columns with default logging
+columns = ['custom_metadata', 'author_email']
+api.portal.add_catalog_metadata(columns_to_add=columns)
+
+# Add columns with custom logger
+import logging
+custom_logger = logging.getLogger('my.package')
+api.portal.add_catalog_metadata(
+    columns_to_add=['publication_date'],
+    logger=custom_logger
+)
+```
+
+% invisible-code-block: python
+%
+% # Verify the columns were added to the catalog
+% catalog = api.portal.get_tool('portal_catalog')
+% self.assertIn('custom_metadata', catalog.schema())
+% self.assertIn('author_email', catalog.schema())
+% self.assertIn('publication_date', catalog.schema())
+
+
 ## Further reading
 
 For more information on possible flags and usage options please see the full {ref}`plone-api-portal` specification.

news/404.feature

@@ -0,0 +1,3 @@
+Added two new helper methods to plone.api.portal:
+- add_catalog_indexes: Adds the specified indexes to portal_catalog if they don't already exist @rohnsha0
+- add_catalog_metadata: Adds the specified metadata columns to portal_catalog if they don't already exist @rohnsha0

setup.py

@@ -28,6 +28,7 @@
     python_requires=">=3.8",
     install_requires=[
         "Acquisition",
+        "Products.ZCTextIndex",
         "Products.statusmessages",
         "Products.PlonePAS",
         "Products.CMFPlone",

src/plone/api/portal.py

@@ -22,6 +22,7 @@
 from zope.schema.interfaces import IVocabularyFactory
 
 import datetime as dtime
+import logging
 import re
 
 
@@ -472,3 +473,101 @@ def get_vocabulary_names():
     :Example: :ref:`portal-get-all-vocabulary-names-example`
     """
     return sorted([name for name, vocabulary in getUtilitiesFor(IVocabularyFactory)])
+
+
+@required_parameters("indexes_to_add")
+def add_catalog_indexes(indexes_to_add, reindex=True, logger=None):
+    """Add the specified indexes to portal_catalog if they don't already exist.
+
+    :param indexes_to_add: [required] List of tuples in format (index_name, index_type)
+    :type indexes_to_add: list
+    :param reindex: Boolean indicating if newly added indexes should be reindexed
+    :type reindex: bool
+    :param logger: Optional logger instance
+    :type logger: logging.Logger
+    :returns: List of newly added index names
+    :rtype: list
+    :Example: :ref:`portal-add-catalog-indexes-example`
+
+    Note: ZCTextIndex indexes require special handling with additional parameters.
+    The function automatically configures lexicon_id, index_type, and doc_attr
+    parameters when adding a ZCTextIndex and creates a minimal lexicon if needed.
+    """
+    if logger is None:
+        logger = logging.getLogger("plone.api.portal")
+
+    catalog = get_tool("portal_catalog")
+    existing_indexes = catalog.indexes()
+    added_indexes = []
+
+    # Import required classes for ZCTextIndex
+    from Products.ZCTextIndex.Lexicon import Lexicon
+
+    for name, meta_type in indexes_to_add:
+        if name not in existing_indexes:
+            if meta_type == "ZCTextIndex":
+                # Ensure a proper configuration for ZCTextIndex
+                extra = {
+                    "lexicon_id": "plone_lexicon",
+                    "index_type": "Okapi BM25 Rank",
+                    "doc_attr": name,
+                }
+
+                # Try to get the existing lexicon or create a minimal one
+                try:
+                    # Try to find the lexicon in the catalog
+                    lexicon = getattr(catalog, "plone_lexicon", None)
+
+                    # If lexicon doesn't exist, create a minimal one
+                    if lexicon is None:
+                        from Products.ZCTextIndex.ZCTextIndex import PLexicon
+
+                        lexicon = PLexicon("plone_lexicon", "Plone Lexicon", Lexicon())
+                        catalog._setObject("plone_lexicon", lexicon)
+
+                    # Add the index with the extra parameters
+                    catalog.addIndex(name, meta_type, extra)
+
+                except Exception as e:
+                    logger.error(f"Error adding ZCTextIndex {name}: {str(e)}")
+                    continue
+            else:
+                # For non-ZCTextIndex types, use standard addIndex
+                catalog.addIndex(name, meta_type)
+
+            added_indexes.append(name)
+            logger.info("Added %s index for field %s.", meta_type, name)
+
+    if reindex and added_indexes:
+        logger.info("Reindexing new indexes: %s", ", ".join(added_indexes))
+        catalog.manage_reindexIndex(ids=added_indexes)
+
+    return added_indexes
+
+
+@required_parameters("columns_to_add")
+def add_catalog_metadata(columns_to_add, logger=None):
+    """Add the specified metadata columns to portal_catalog.
+
+    :param columns_to_add: [required] List of column names to add
+    :type columns_to_add: list
+    :param logger: Optional custom logger instance
+    :type logger: logging.Logger
+    :returns: List of names of columns that were added
+    :rtype: list
+    :Example: :ref:`portal-add-catalog-metadata-example`
+    """
+    if logger is None:
+        logger = logging.getLogger("plone.api.portal")
+
+    catalog = get_tool("portal_catalog")
+    existing_columns = catalog.schema()
+
+    added_columns = []
+    for name in columns_to_add:
+        if name not in existing_columns:
+            catalog.addColumn(name)
+            added_columns.append(name)
+            logger.info("Added metadata column: %s", name)
+
+    return added_columns

src/plone/api/tests/test_portal.py

@@ -963,3 +963,128 @@ def test_vocabulary_terms(self):
         states = [term.value for term in states_vocabulary]
         self.assertIn("private", states)
         self.assertIn("published", states)
+
+    def test_add_catalog_indexes(self):
+        """Test adding catalog indexes."""
+        import logging
+
+        catalog = portal.get_tool("portal_catalog")
+
+        # Test adding new indexes
+        test_indexes = [
+            ("test_field1", "FieldIndex"),
+            ("test_field2", "KeywordIndex"),
+        ]
+
+        added = portal.add_catalog_indexes(test_indexes)
+
+        # Verify indexes were added
+        self.assertEqual(len(added), 2)
+        self.assertIn("test_field1", added)
+        self.assertIn("test_field2", added)
+        self.assertIn("test_field1", catalog.indexes())
+        self.assertIn("test_field2", catalog.indexes())
+
+        # Test adding already existing indexes
+        added = portal.add_catalog_indexes(test_indexes)
+        self.assertEqual(len(added), 0)  # No new indexes should be added
+
+        # Test with reindex=False
+        test_indexes2 = [
+            ("test_field3", "FieldIndex"),
+        ]
+
+        # Create a mock for catalog.manage_reindexIndex to verify it's called or not
+        original_reindex = catalog.manage_reindexIndex
+
+        try:
+            reindex_called = [False]
+
+            def mock_reindex(ids=None):
+                reindex_called[0] = True
+                self.assertEqual(ids, ["test_field3"])
+                original_reindex(ids)
+
+            catalog.manage_reindexIndex = mock_reindex
+
+            portal.add_catalog_indexes(test_indexes2, reindex=True)
+            self.assertTrue(reindex_called[0])
+
+            # Reset flag and test with reindex=False
+            reindex_called[0] = False
+            portal.add_catalog_indexes([("test_field4", "FieldIndex")], reindex=False)
+            self.assertFalse(reindex_called[0])
+
+        finally:
+            # Restore original method
+            catalog.manage_reindexIndex = original_reindex
+
+        # Test with custom logger
+        test_logger = logging.getLogger("test.plone.api.portal")
+
+        with self.assertLogs("test.plone.api.portal", level="INFO") as cm:
+            portal.add_catalog_indexes(
+                [("test_field5", "FieldIndex")], logger=test_logger
+            )
+
+        log_output = "\n".join(cm.output)
+        self.assertIn("Added FieldIndex index for field test_field5", log_output)
+        self.assertIn("Reindexing new indexes: test_field5", log_output)
+
+    def test_add_catalog_indexes_zctext_index(self):
+        """Test adding a ZCTextIndex type index with appropriate extra parameters."""
+        from plone.api import portal
+
+        # Mock the catalog
+        catalog_mock = mock.Mock()
+        catalog_mock.indexes = mock.Mock(return_value=[])
+
+        # Replace the get_tool function to return our mock
+        with mock.patch.object(portal, "get_tool", return_value=catalog_mock):
+            # Call the function with a ZCTextIndex
+            portal.add_catalog_indexes([("myindex", "ZCTextIndex")], reindex=False)
+
+            # Verify that addIndex was called with the correct extra parameters
+            catalog_mock.addIndex.assert_called_once()
+            name, meta_type, extra = catalog_mock.addIndex.call_args[0]
+            self.assertEqual(name, "myindex")
+            self.assertEqual(meta_type, "ZCTextIndex")
+            self.assertEqual(extra["lexicon_id"], "plone_lexicon")
+            self.assertEqual(extra["index_type"], "Okapi BM25 Rank")
+            self.assertEqual(extra["doc_attr"], "myindex")
+
+            # Verify that manage_reindexIndex wasn't called (reindex=False)
+            catalog_mock.manage_reindexIndex.assert_not_called()
+
+    def test_add_catalog_metadata(self):
+        """Test adding catalog metadata columns."""
+        from plone.api.portal import add_catalog_metadata
+
+        import logging
+
+        catalog = portal.get_tool("portal_catalog")
+
+        # Test adding new columns
+        test_columns = ["test_col1", "test_col2"]
+
+        added = add_catalog_metadata(test_columns)
+
+        # Verify columns were added
+        self.assertEqual(len(added), 2)
+        self.assertIn("test_col1", added)
+        self.assertIn("test_col2", added)
+        self.assertIn("test_col1", catalog.schema())
+        self.assertIn("test_col2", catalog.schema())
+
+        # Test adding already existing columns
+        added = add_catalog_metadata(test_columns)
+        self.assertEqual(len(added), 0)  # No new columns should be added
+
+        # Test with custom logger
+        test_logger = logging.getLogger("test.plone.api.portal")
+
+        with self.assertLogs("test.plone.api.portal", level="INFO") as cm:
+            add_catalog_metadata(["test_col3"], logger=test_logger)
+
+        log_output = "\n".join(cm.output)
+        self.assertIn("Added metadata column: test_col3", log_output)