Add a method Client.space_info() which returns information about the types and number of instances in a space.

Author: apdavison

fairgraph/client.py

@@ -34,6 +34,7 @@
     have_kg_core = False
 
 from .errors import AuthenticationError, AuthorizationError, ResourceExistsError
+from .registry import lookup_type
 
 if TYPE_CHECKING:
     from .kgobject import KGObject
@@ -583,7 +584,10 @@ def configure_space(self, space_name: Optional[str] = None, types: Optional[List
                 space_name = f"collab-{collab_id}"
         result = self._kg_admin_client.create_space_definition(space=space_name)
         if result:  # error
-            raise Exception(f"Unable to configure KG space for space '{space_name}': {result}")
+            err_msg = f"Unable to configure KG space for space '{space_name}': {result}"
+            if not space_name.startswith("collab="):
+                err_msg += f". If you are trying to configure a collab space, ensure the space name starts with 'collab-'"
+            raise Exception(err_msg)
         for cls in types:
             result = self._kg_admin_client.assign_type_to_space(space=space_name, target_type=cls.type_)
             if result:  # error
@@ -601,6 +605,29 @@ def move_to_space(self, uri: str, destination_space: str):
         if response.error:
             raise Exception(response.error)
 
+    def space_info(self, space_name: str, scope: str = "released"):
+        """
+        Return information about the types and number of instances in a space.
+
+        The return format is a dictionary whose keys are classes and the values are the
+        number of instances of each class in the given spaces.
+        """
+        result = self._kg_client.types.list(space=space_name, stage=STAGE_MAP[scope])
+        if result.error:
+            raise Exception(result.error)
+        response = {}
+        for item in result.data:
+            try:
+                cls = lookup_type(item.identifier)
+            except KeyError as err:
+                if "vocab/meta/type/Query" in str(err):
+                    pass
+                else:
+                    raise
+            else:
+                response[cls] = item.occurrences
+        return response
+
     def is_released(self, uri: str, with_children: bool = False) -> bool:
         """
         Release status of a KG instance identified by its URI.