Add datasources namespace API

src/ansys/dpf/core/data_sources.py

@@ -680,6 +680,18 @@ def register_namespace(self, result_key: str, namespace: str):
         """
         self._api.data_sources_register_namespace(self, result_key, namespace)
 
+    def namespace(self, result_key: str) -> str:
+        """
+        Return the namespace associated to a result_key. The namespace identifies to which operator plugin a call should be delegated to.
+
+        Parameters
+        ----------
+        result_key:
+            Extension of the file, which is used as a key for choosing the correct
+            plugin when a result is requested by an operator.
+        """
+        return self._api.data_sources_get_namespace(self, result_key)
+
     def __str__(self):
         """Describe the entity.
 

src/ansys/dpf/gate/data_sources_grpcapi.py

@@ -176,3 +176,8 @@ def data_sources_get_key(dataSources, index, num_path):
     def data_sources_get_path(dataSources, key, index):
         response = _get_stub(dataSources._server).List(dataSources._internal_obj)
         return list(response.paths[key].paths)[index]
+
+    @staticmethod
+    def data_sources_get_namespace(dataSources, key):
+        response = _get_stub(dataSources._server).List(dataSources._internal_obj)
+        return response.namespaces[key]

tests/test_datasources.py

@@ -25,6 +25,7 @@
 import pytest
 
 from ansys import dpf
+from ansys.dpf.core import examples
 import conftest
 
 skip_always = pytest.mark.skipif(True, reason="Investigate why this is failing")
@@ -173,3 +174,13 @@ def test_register_namespace(allkindofcomplexity, server_type):
     with pytest.raises(Exception):
         op = dpf.core.operators.result.displacement(data_sources=data_sources, server=server_type)
         assert op.eval() is not None
+
+
+def test_namespace(allkindofcomplexity, server_type):
+    data_sources = dpf.core.DataSources(allkindofcomplexity, server=server_type)
+    assert data_sources.namespace(data_sources.result_key) == "mapdl"
+
+    cas_h5_file = examples.download_fluent_axial_comp(server=server_type)["cas"][0]
+    data_sources = dpf.core.DataSources(server=server_type)
+    data_sources.set_result_file_path(cas_h5_file)
+    assert data_sources.namespace(data_sources.result_key) == "cff"