Add endpoint to access block_types indexer

README.md

@@ -111,6 +111,47 @@ By default it is only added for new Plone sites.
 To add it to an existing site, run `plone.volto.upgrades.add_block_types_index` manually.
 
 
+### Block types endpoint
+
+`plone.volto` exposes the `block_types` index via the `/@blocktypes` REST API endpoint.
+It can be used to get information about which and how many blocks are used where.
+
+```py
+import requests
+requests.get("http://localhost:8080/Plone/@blocktypes")
+```
+
+```json
+{
+  "title": 3,
+  "slate": 18,
+  "teaser": 6,
+}
+```
+
+It can also be used to get detailed information where a block is used and how often:
+
+```py
+import requests
+requests.get("http://localhost:8080/Plone/@blocktypes/teaser")
+```
+
+```json
+[
+  {
+    "@id": "http://localhost:8080/Plone",
+    "title": "Website",
+    "count": 2,
+  },
+  {
+    "@id": "http://localhost:8080/Plone/lorem-ipsum",
+    "title": "Lorem Ipsum",
+    "count": 4,
+  }
+]
+```
+
+
 ### Multilingual support
 
 `plone.volto` supports multilingual websites.

news/+block_types_count.feature

@@ -0,0 +1 @@
+Added a `block_types` metadata column to the catalog to include a count for each type. @jnptk

news/+block_types_service.feature

@@ -0,0 +1 @@
+Add `/@blocktypes` endpoint to expose `block_types` index. @jnptk

src/plone/volto/configure.zcml

@@ -10,11 +10,18 @@
 
   <i18n:registerTranslations directory="locales" />
 
+  <include
+      package="Products.CMFCore"
+      file="permissions.zcml"
+      />
+
   <include file="dependencies.zcml" />
+  <include file="permissions.zcml" />
 
   <include package=".behaviors" />
   <include package=".browser" />
   <include package=".indexers" />
+  <include package=".services" />
 
   <include file="profiles.zcml" />
   <include file="patches.zcml" />

src/plone/volto/indexers/indexers.py

@@ -4,6 +4,8 @@
 from plone.restapi.blocks import visit_blocks
 from plone.volto.behaviors.preview import IPreview
 
+import collections
+
 
 @indexer(IPreview)
 def hasPreviewImage(obj):
@@ -40,9 +42,9 @@ def image_field_indexer(obj):
 def block_types_indexer(obj):
     """Indexer for all block types included in a page."""
     obj = aq_base(obj)
-    block_types = set()
+    block_types = collections.Counter()
     for block in visit_blocks(obj, obj.blocks):
         block_type = block.get("@type")
         if block_type:
-            block_types.add(block_type)
+            block_types[block_type] += 1
     return block_types

src/plone/volto/permissions.zcml

@@ -0,0 +1,8 @@
+<configure xmlns="http://namespaces.zope.org/zope">
+
+  <permission
+      id="plone.volto.service.BlockTypes"
+      title="Plone Site Setup: Block Types"
+      />
+
+</configure>

src/plone/volto/profiles/default/catalog.xml

@@ -4,6 +4,7 @@
   <column value="head_title" />
   <column value="hasPreviewImage" />
   <column value="image_field" />
+  <column value="block_types" />
   <index meta_type="KeywordIndex"
          name="block_types"
   >

src/plone/volto/profiles/default/rolemap.xml

@@ -0,0 +1,11 @@
+<?xml version="1.0" encoding="utf-8"?>
+<rolemap>
+  <permissions>
+    <permission acquire="True"
+                name="Plone Site Setup: Block Types"
+    >
+      <role name="Manager" />
+      <role name="Site Administrator" />
+    </permission>
+  </permissions>
+</rolemap>

src/plone/volto/services/blocktypes/configure.zcml

@@ -0,0 +1,15 @@
+<configure
+    xmlns="http://namespaces.zope.org/zope"
+    xmlns:plone="http://namespaces.plone.org/plone"
+    xmlns:zcml="http://namespaces.zope.org/zcml"
+    >
+
+  <plone:service
+      method="GET"
+      factory=".get.BlockTypesGet"
+      for="zope.interface.Interface"
+      permission="plone.volto.service.BlockTypes"
+      name="@blocktypes"
+      />
+
+</configure>

src/plone/volto/services/blocktypes/get.py

@@ -0,0 +1,56 @@
+from plone import api
+from plone.restapi.behaviors import IBlocks
+from plone.restapi.services import Service
+from zope.interface import implementer
+from zope.publisher.interfaces import IPublishTraverse
+
+import collections
+
+
+@implementer(IPublishTraverse)
+class BlockTypesGet(Service):
+    def __init__(self, context, request):
+        super().__init__(context, request)
+        self.block_type = None
+
+    def publishTraverse(self, request, name):
+        self.block_type = name
+        return self
+
+    def reply(self):
+        catalog = api.portal.get_tool(name="portal_catalog")
+        request_body = self.request.form
+        type = self.block_type
+
+        query = {
+            "object_provides": IBlocks.__identifier__,
+        }
+
+        if request_body.get("path"):
+            query["path"] = request_body["path"]
+
+        if type:
+            result = []
+            query["block_types"] = self.block_type
+            brains = catalog.searchResults(**query)
+
+            for brain in brains:
+                result.append(
+                    {
+                        "@id": brain.getURL(),
+                        "title": brain.Title,
+                        "count": brain.block_types[self.block_type],
+                    }
+                )
+        else:
+            result = {}
+            brains = catalog.searchResults(**query)
+            block_types_total = collections.Counter()
+
+            for brain in brains:
+                block_types_total.update(brain.block_types)
+
+            for block_type, count in block_types_total.items():
+                result[block_type] = count
+
+        return result

src/plone/volto/services/configure.zcml

@@ -0,0 +1,8 @@
+<configure
+    xmlns="http://namespaces.zope.org/zope"
+    xmlns:zcml="http://namespaces.zope.org/zcml"
+    >
+
+  <include package=".blocktypes" />
+
+</configure>

src/plone/volto/upgrades.py

@@ -199,3 +199,18 @@ def add_large_image_scales(context):
     if not any(item.startswith("4k ") for item in value):
         value.append("4k 4000:65536")
     api.portal.set_registry_record("plone.allowed_sizes", value)
+
+
+def reindex_block_types(context):
+    catalog = getToolByName(context, "portal_catalog")
+    brains = catalog(object_provides="plone.restapi.behaviors.IBlocks")
+    total = len(brains)
+    for index, brain in enumerate(brains):
+        obj = brain.getObject()
+        obj.reindexObject(idxs=["block_types"], update_metadata=1)
+        logger.info(f"Reindexing object {brain.getPath()}.")
+        if index % 250 == 0:
+            logger.info(f"Reindexed {index}/{total} objects")
+            transaction.commit()
+    logger.info(f"Reindexed {total} objects")
+    transaction.commit()

src/plone/volto/upgrades.zcml

@@ -107,4 +107,19 @@
         handler=".upgrades.add_large_image_scales"
         />
   </genericsetup:upgradeSteps>
+
+  <genericsetup:upgradeSteps
+      profile="plone.volto:default"
+      source="1100"
+      destination="1101"
+      >
+    <genericsetup:upgradeDepends
+        title="Add block_types metadata column and apply rolemap"
+        import_steps="catalog rolemap"
+        />
+    <genericsetup:upgradeStep
+        title="Reindex to populate the block_types metadata"
+        handler=".upgrades.reindex_block_types"
+        />
+  </genericsetup:upgradeSteps>
 </configure>

tests/services/blocktypes/test_blocktypes_get.py

@@ -0,0 +1,56 @@
+from plone import api
+
+import pytest
+import transaction
+
+
+@pytest.fixture(scope="class")
+def portal(portal_class):
+    yield portal_class
+
+
+@pytest.fixture(scope="class")
+def contents(portal):
+    with api.env.adopt_roles(["Manager"]):
+        doc = api.content.create(portal, type="Document", id="lorem-ipsum")
+        doc.blocks = {
+            "1": {"@type": "title"},
+            "2": {"@type": "teaser"},
+            "3": {
+                "@type": "gridBlock",
+                "blocks": {
+                    "1": {"@type": "teaser"},
+                    "2": {"@type": "teaser"},
+                    "3": {"@type": "teaser"},
+                },
+            },
+        }
+        doc.reindexObject(idxs=["block_types"])
+        transaction.commit()
+
+
+class TestBlockTypesGet:
+    @pytest.fixture(autouse=True)
+    def _setup(self, contents, portal, api_manager_request):
+        self.portal = portal
+        self.api_session = api_manager_request
+
+    def test_response_type(self):
+        response = self.api_session.get("/@blocktypes")
+        data = response.json()
+        assert isinstance(data, dict)
+
+    def test_response_type_with_id(self):
+        response = self.api_session.get("/@blocktypes/title")
+        data = response.json()
+        assert isinstance(data, list)
+
+    def test_filtered(self):
+        response = self.api_session.get("/@blocktypes?path=/plone/lorem-ipsum")
+        data = response.json()
+        assert len(data) == 3
+
+    def test_filtered_with_id(self):
+        response = self.api_session.get("/@blocktypes/teaser?path=/plone/lorem-ipsum")
+        data = response.json()
+        assert len(data) == 1

tests/services/conftest.py

@@ -0,0 +1,52 @@
+from plone.app.testing import SITE_OWNER_NAME
+from plone.app.testing import SITE_OWNER_PASSWORD
+from plone.restapi.testing import RelativeSession
+from zope.component.hooks import site
+
+import pytest
+
+
+@pytest.fixture()
+def portal(functional):
+    """Provide the Plone portal instance for functional tests."""
+    yield functional["portal"]
+
+
+@pytest.fixture(scope="class")
+def portal_class(functional_class):
+    """Provide the Plone portal instance for class-scoped functional tests."""
+    if hasattr(functional_class, "testSetUp"):
+        functional_class.testSetUp()
+    portal = functional_class["portal"]
+    with site(portal):
+        yield portal
+    if hasattr(functional_class, "testTearDown"):
+        functional_class.testTearDown()
+
+
+@pytest.fixture()
+def request_api_factory(portal):
+    """Provide a factory function for creating Plone REST API session objects."""
+
+    def factory():
+        url = portal.absolute_url()
+        api_session = RelativeSession(f"{url}/++api++")
+        return api_session
+
+    return factory
+
+
+@pytest.fixture()
+def api_anon_request(request_api_factory):
+    """Provide an unauthenticated REST API session for anonymous requests."""
+    request = request_api_factory()
+    yield request
+
+
+@pytest.fixture()
+def api_manager_request(request_api_factory):
+    """Provide an authenticated REST API session with manager privileges."""
+    request = request_api_factory()
+    request.auth = (SITE_OWNER_NAME, SITE_OWNER_PASSWORD)
+    yield request
+    request.auth = ()