FEAT - 328 - Add support for maximum specification link depth

examples/3_Advanced_Topics/3-3_Database-specific_configuration.py

@@ -69,3 +69,17 @@
 spec_query = spec_query.with_batch_size(5)
 spec_query
 # -
+
+# ## Specification to Specification links
+
+# This package supports traversing specification to specification links, by default all such links will be followed.
+# For typical databases this is the correct and desired behaviour, however in some circumstances this may cause query
+# times and response sizes to become very large. In such cases you can control the number of nested links that will be
+# followed using the ``maximum_spec_link_depth`` parameter on the ``BomAnalyticsClient`` object.
+
+# The default value is None, setting it to a positive integer will limit the depth to at most that many nested links.
+
+# + tags=[]
+cxn.maximum_spec_link_depth = 2
+cxn
+# -

pyproject.toml

@@ -41,7 +41,7 @@ python = "^3.7.1"
 # Packages for core library
 importlib_metadata = { version = ">=1.0", python  = "<3.8" }  # Granta MI STK requires 3.4.0
 ansys-openapi-common = "< 2.0.0"
-ansys-grantami-bomanalytics-openapi = "1.0.0"
+ansys-grantami-bomanalytics-openapi = "1.1.0"
 
 # Packages for the examples extra
 jupyterlab = { version = "3.6.5", optional = true }

src/ansys/grantami/bomanalytics/_connection.py

@@ -26,7 +26,7 @@
     Identifier used internally by the Granta MI Server.
 """
 
-from typing import overload, TYPE_CHECKING, Union, Dict, Optional, Type, Any
+from typing import overload, TYPE_CHECKING, Union, Dict, Optional, Type, Any, Tuple, List
 
 from ansys.openapi.common import (  # type: ignore[import]
     ApiClientFactory,
@@ -193,15 +193,15 @@ class BomAnalyticsClient(ApiClient):
     """
 
     def __init__(self, servicelayer_url: str, **kwargs: Any) -> None:
-        self._sl_url = servicelayer_url.strip("/")
+        self._sl_url: str = servicelayer_url.strip("/")
         sl_url_with_service = self._sl_url + SERVICE_PATH
         logger.debug("Creating BomAnalyticsClient")
         logger.debug(f"Base Service Layer URL: {self._sl_url}")
         logger.debug(f"Service URL: {sl_url_with_service}")
 
         super().__init__(api_url=sl_url_with_service, **kwargs)
 
-        self._db_key = DEFAULT_DBKEY
+        self._db_key: str = DEFAULT_DBKEY
         self._table_names: Dict[str, Optional[str]] = {
             "material_universe_table_name": None,
             "inhouse_materials_table_name": None,
@@ -210,14 +210,47 @@ def __init__(self, servicelayer_url: str, **kwargs: Any) -> None:
             "substances_table_name": None,
             "coatings_table_name": None,
         }
+        self._max_spec_depth: Optional[int] = None
 
     def __repr__(self) -> str:
-        base_repr = f'<BomServicesClient: url="{self._sl_url}", dbkey="{self._db_key}"'
-        custom_tables = ", ".join([f'{k}="{v}"' for k, v in self._table_names.items() if v])
-        if custom_tables:
-            return base_repr + f", {custom_tables}>"
-        else:
-            return base_repr + ">"
+        max_link_value: Union[str, int] = (
+            "unlimited" if self.maximum_spec_link_depth is None else self.maximum_spec_link_depth
+        )
+        repr_entries: List[Tuple[str, Union[str, int, None]]] = [
+            ("url", self._sl_url),
+            ("maximum_spec_link_depth", max_link_value),
+            ("dbkey", self._db_key),
+        ]
+        for k, v in self._table_names.items():
+            if v:
+                repr_entries.append((k, v))
+        rendered_entries = []
+        for name, value in repr_entries:
+            if isinstance(value, str):
+                value = f'"{value}"'
+            rendered_entries.append(f"{name}={value}")
+        return f'<BomServicesClient: {", ".join(rendered_entries)}>'
+
+    @property
+    def maximum_spec_link_depth(self) -> Optional[int]:
+        """Limits the maximum depth of specification to specification links that will be followed when determining
+        impacted substances or compliance.
+        If None then no limit will be applied, this may lead to performance issues on databases with large numbers of
+        these links.
+
+        Returns
+        -------
+        Optional[int]
+            Maximum depth of specification to specification links that will be followed.
+
+        """
+        return self._max_spec_depth
+
+    @maximum_spec_link_depth.setter
+    def maximum_spec_link_depth(self, value: Optional[int]) -> None:
+        if value is not None and value < 0:
+            raise ValueError("maximum_spec_link_depth must be a non-negative integer or None")
+        self._max_spec_depth = value
 
     def set_database_details(
         self,
@@ -382,14 +415,20 @@ def _query_arguments(
         The database key is always required. The default is only included here for convenience.
         """
 
+        config = models.CommonRequestConfig()
+        if self._max_spec_depth is not None:
+            logger.info(f"Using maximum spec to spec link depth: {self._max_spec_depth}")
+            config.maximum_spec_to_spec_link_depth = self._max_spec_depth + 1
+        else:
+            logger.info(f"Using no spec to spec link depth limit, all links will be followed")
         if any(self._table_names.values()):
-            config = models.CommonRequestConfig(**self._table_names)
+            for table_type, name in self._table_names.items():
+                setattr(config, table_type, name)
             table_mapping = [f"{n}: {v}" for n, v in self._table_names.items() if v]
             logger.info(f"Using custom table config:")
             for line in table_mapping:
                 logger.info(line)
         else:
-            config = None
             logger.info(f"Using default table config")
 
         if self._db_key != DEFAULT_DBKEY:

tests/conftest.py

@@ -16,12 +16,18 @@
 
 @pytest.fixture(scope="session")
 def default_connection():
-    connection = Connection(api_url=sl_url).with_credentials(read_username, read_password).connect()
+    if read_username is not None:
+        connection = Connection(api_url=sl_url).with_credentials(read_username, read_password).connect()
+    else:
+        connection = Connection(api_url=sl_url).with_autologon().connect()
     return connection
 
 
 def _get_connection(request, url, username, password):
-    connection = Connection(api_url=url).with_credentials(username, password).connect()
+    if username is not None:
+        connection = Connection(api_url=url).with_credentials(username, password).connect()
+    else:
+        connection = Connection(api_url=url).with_autologon().connect()
     if request.param:
         if isinstance(request.param, str):
             db_key = request.param

tests/test_connection.py

@@ -30,30 +30,47 @@ def test_default_dbkey(mock_connection):
 
 
 def test_repr_default_dbkey(mock_connection):
-    assert repr(mock_connection) == f'<BomServicesClient: url="{SL_URL}", ' 'dbkey="MI_Restricted_Substances">'
+    assert (
+        repr(mock_connection)
+        == f'<BomServicesClient: url="{SL_URL}", maximum_spec_link_depth="unlimited", dbkey="MI_Restricted_Substances">'
+    )
 
 
 def test_repr_custom_dbkey(mock_connection):
     mock_connection.set_database_details(database_key="RS_DB")
-    assert repr(mock_connection) == f'<BomServicesClient: url="{SL_URL}", dbkey="RS_DB">'
+    assert (
+        repr(mock_connection)
+        == f'<BomServicesClient: url="{SL_URL}", maximum_spec_link_depth="unlimited", dbkey="RS_DB">'
+    )
 
 
 def test_repr_default_dbkey_custom_table(mock_connection):
     mock_connection.set_database_details(specifications_table_name="My Specs")
     assert (
-        repr(mock_connection) == f'<BomServicesClient: url="{SL_URL}", '
+        repr(mock_connection) == f'<BomServicesClient: url="{SL_URL}", maximum_spec_link_depth="unlimited", '
         'dbkey="MI_Restricted_Substances", specifications_table_name="My Specs">'
     )
 
 
 def test_repr_custom_dbkey_custom_table(mock_connection):
     mock_connection.set_database_details(database_key="RS_DB", specifications_table_name="My Specs")
     assert (
-        repr(mock_connection) == f'<BomServicesClient: url="{SL_URL}", '
+        repr(mock_connection) == f'<BomServicesClient: url="{SL_URL}", maximum_spec_link_depth="unlimited", '
         'dbkey="RS_DB", specifications_table_name="My Specs">'
     )
 
 
+@pytest.mark.parametrize("value", [None, 0, 1, 3000])
+def test_set_max_spec_link_depth_with_valid_inputs(mock_connection, value):
+    mock_connection.maximum_spec_link_depth = value
+    assert mock_connection.maximum_spec_link_depth == value
+
+
+def test_set_max_spec_link_depth_with_invalid_input(mock_connection):
+    with pytest.raises(ValueError, match="maximum_spec_link_depth must be a non-negative integer or None"):
+        mock_connection.maximum_spec_link_depth = -1
+
+
 class TestConnectToSL:
     @pytest.mark.parametrize(
         "sl_url", ["http://host/path/", "http://host/path", "https://host/path/", "https://host/path"]

tests/test_integration_tests.py

@@ -1,7 +1,7 @@
 import pytest
 from .inputs import sample_bom_complex, sample_bom_custom_db
 from ansys.grantami.bomanalytics import queries, GrantaMIException
-from .common import LEGISLATIONS, INDICATORS
+from .common import LEGISLATIONS, INDICATORS, CUSTOM_TABLES
 
 pytestmark = pytest.mark.integration
 
@@ -155,3 +155,51 @@ def test_withdrawn_records_return_warning_messages_if_not_acting_as_read(self, c
         assert any(
             "has 1 substance row(s) having more than one linked substance. " in msg.message for msg in results.messages
         )
+
+
+class TestSpecLinkDepth:
+    spec_ids = ["MIL-DTL-53039,TypeII"]
+    legislation_ids = ["EU REACH - The Candidate List"]
+
+    @pytest.fixture(scope="class")
+    def connection_with_custom_tables(self, default_connection):
+        db_key = "MI_Restricted_Substances_Custom_Tables"
+        base_db_key = default_connection._db_key
+        default_connection.set_database_details(database_key=db_key, **{pn: tn for pn, tn in CUSTOM_TABLES})
+        yield default_connection
+        default_connection.set_database_details(database_key=base_db_key, **{pn: None for pn, _ in CUSTOM_TABLES})
+
+    @pytest.fixture(scope="function")
+    def connection(self, connection_with_custom_tables):
+        old_depth = connection_with_custom_tables.maximum_spec_link_depth
+        yield connection_with_custom_tables
+        connection_with_custom_tables.maximum_spec_link_depth = old_depth
+
+    def test_legislation_is_affected_with_link_depth_one(self, connection):
+        connection.maximum_spec_link_depth = 1
+
+        query = (
+            queries.SpecificationImpactedSubstancesQuery()
+            .with_specification_ids(self.spec_ids)
+            .with_legislations(self.legislation_ids)
+        )
+        response = connection.run(query)
+        assert len(response.impacted_substances) == 1
+        assert response.impacted_substances[0].cas_number == "872-50-4"
+        assert len(response.impacted_substances_by_legislation) == 1
+        legislation_name = self.legislation_ids[0]
+        assert legislation_name in response.impacted_substances_by_legislation
+        impacted_by_reach = response.impacted_substances_by_legislation[legislation_name]
+        assert len(impacted_by_reach) == 1
+        assert impacted_by_reach[0].cas_number == "872-50-4"
+
+    def test_legislation_is_not_affected_with_no_links(self, connection):
+        connection.maximum_spec_link_depth = 0
+
+        query = (
+            queries.SpecificationImpactedSubstancesQuery()
+            .with_specification_ids(self.spec_ids)
+            .with_legislations(self.legislation_ids)
+        )
+        response = connection.run(query)
+        assert len(response.impacted_substances) == 0