add get_children_by_path()

asyncua/common/node.py

@@ -508,6 +508,39 @@ async def get_child(self, path, return_all=False):
             return [Node(self.session, target.TargetId) for target in result.Targets]
         return Node(self.session, result.Targets[0].TargetId)
 
+    async def get_children_by_path(self, paths, raise_on_partial_error=True):
+        """
+        get children specified by their paths from this node.
+        A path might be:
+        * a string representing a qualified name.
+        * a qualified name
+        * a list of string
+        * a list of qualified names
+        """
+        bpaths = []
+        for path in paths:
+            if type(path) not in (list, tuple):
+                path = [path]
+            rpath = self._make_relative_path(path)
+            bpath = ua.BrowsePath()
+            bpath.StartingNode = self.nodeid
+            bpath.RelativePath = rpath
+            bpaths.append(bpath)
+
+        results = await self.session.translate_browsepaths_to_nodeids(bpaths)
+        try:
+            if raise_on_partial_error:
+                for result in results:
+                    result.StatusCode.check()
+        except ua.UaStatusCodeError:
+            codes = [result.StatusCode.value for result in results]
+            raise ua.UaStatusCodeErrors(codes)
+        return [
+            [Node(self.session, target.TargetId) for target in result.Targets]
+            if result.StatusCode.is_good() else None
+            for result in results
+        ]
+
     def _make_relative_path(self, path):
         rpath = ua.RelativePath()
         for item in path:

asyncua/sync.py

@@ -506,6 +506,10 @@ def get_user_access_level(self):
     def get_child(self, path):
         pass
 
+    @syncmethod
+    def get_children_by_path(self, paths, raise_on_partial_error=True):
+        pass
+
     @syncmethod
     def read_raw_history(self, starttime=None, endtime=None, numvalues=0, return_bounds=True):
         pass

asyncua/ua/__init__.py

@@ -6,3 +6,4 @@
 from .uaprotocol_auto import *
 from .uaprotocol_hand import *
 from .uatypes import *  # TODO: This should be renamed to uatypes_hand
+from .uaerrors import UaStatusCodeErrors

asyncua/ua/uaerrors/_base.py

@@ -66,6 +66,36 @@ def code(self):
         return self.args[0]
 
 
+class UaStatusCodeErrors(UaStatusCodeError):
+    def __new__(cls, *args):
+        # use the default implementation
+        self = UaError.__new__(cls, *args)
+        return self
+
+    def __init__(self, codes):
+        """
+        :param codes: The codes of the results.
+        """
+        self.codes = codes
+
+    def __str__(self):
+        # import here to avoid circular import problems
+        import asyncua.ua.status_codes as status_codes
+
+        return "[{0}]".format(", ".join(["{1}({0})".format(*status_codes.get_name_and_doc(code)) for code in self.codes]))
+
+    @property
+    def code(self):
+        """
+        The code of the status error.
+        """
+        # import here to avoid circular import problems
+        from asyncua.ua.uatypes import StatusCode
+
+        error_codes = [code for code in self.codes if not StatusCode(code).is_good()]
+        return error_codes[0] if len(error_codes) > 0 else None
+
+
 class UaStringParsingError(UaError):
     pass
 

tests/test_common.py

@@ -399,6 +399,29 @@ async def test_get_node_by_nodeid(opc):
     assert server_time_node == correct
 
 
+async def test_get_children_by_path(opc):
+    root = opc.opc.nodes.root
+    [server_time_nodes] = await root.get_children_by_path([['0:Objects', '0:Server', '0:ServerStatus', '0:CurrentTime']])
+    correct = opc.opc.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_CurrentTime))
+    assert len(server_time_nodes) == 1
+    assert server_time_nodes[0] == correct
+    with pytest.raises(ua.UaStatusCodeError) as e:
+        await root.get_children_by_path([['0:Objects', '0:Server', '0:ServerStatus', '0:CurrentTime'], ['0:Objects', '0:Unknown']])
+    assert e.type == ua.UaStatusCodeErrors
+    assert e.value.codes == [ua.StatusCodes.Good, ua.StatusCodes.BadNoMatch]
+    assert e.value.code == ua.StatusCodes.BadNoMatch
+    [server_time_nodes, unknown] = await root.get_children_by_path(
+        [
+            ['0:Objects', '0:Server', '0:ServerStatus', '0:CurrentTime'],
+            ['0:Objects', '0:Unknown']
+        ],
+        raise_on_partial_error=False
+    )
+    assert len(server_time_nodes) == 1
+    assert server_time_nodes[0] == correct
+    assert unknown is None
+
+
 async def test_datetime_read_value(opc):
     time_node = opc.opc.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_CurrentTime))
     dt = await time_node.read_value()
@@ -556,6 +579,10 @@ async def test_same_browse_name(opc):
     assert len(nodes) == 2
     assert nodes[0] == v
     assert nodes[1] == v2
+    [nodes] = await objects.get_children_by_path([['2:MyBNameFolder', '2:MyBName', '2:MyBNameTarget']])
+    assert len(nodes) == 2
+    assert nodes[0] == v
+    assert nodes[1] == v2
     await opc.opc.delete_nodes([f, o, o2, v, v2])