data: add support of anydata into dict_to_dnode

This patch adds support of anydata within dict_to_dnode. To do that new
options were introduced into dict_to_dnode() API

Signed-off-by: Stefan Gula <steweg@gmail.com>

Author: steweg

cffi/cdefs.h

@@ -990,6 +990,7 @@ struct lyd_node_any {
 };
 
 LY_ERR lyd_any_value_str(const struct lyd_node *, char **);
+LY_ERR lyd_new_any(struct lyd_node *, const struct lys_module *, const char *, const void *, LYD_ANYDATA_VALUETYPE, uint32_t, struct lyd_node **);
 
 #define LYD_MERGE_DEFAULTS ...
 #define LYD_MERGE_DESTRUCT ...

libyang/data.py

@@ -2,13 +2,15 @@
 # Copyright (c) 2021 RACOM s.r.o.
 # SPDX-License-Identifier: MIT
 
+import json
 import logging
 from typing import IO, Any, Dict, Iterator, Optional, Tuple, Union
 
 from _libyang import ffi, lib
 from .keyed_list import KeyedList
 from .schema import (
     Module,
+    SAnydata,
     SContainer,
     SLeaf,
     SLeafList,
@@ -104,6 +106,21 @@ def newval_flags(
     return flags
 
 
+# -------------------------------------------------------------------------------------
+def anydata_format(fmt_string: str) -> int:
+    if fmt_string == "datatree":
+        return lib.LYD_ANYDATA_DATATREE
+    if fmt_string == "string":
+        return lib.LYD_ANYDATA_STRING
+    if fmt_string == "xml":
+        return lib.LYD_ANYDATA_XML
+    if fmt_string == "json":
+        return lib.LYD_ANYDATA_JSON
+    if fmt_string == "lyb":
+        return lib.LYD_ANYDATA_LYB
+    raise ValueError("unknown anydata format: %r" % fmt_string)
+
+
 # -------------------------------------------------------------------------------------
 def parser_flags(
     lyb_mod_update: bool = False,
@@ -1147,6 +1164,8 @@ def dict_to_dnode(
     rpc: bool = False,
     rpcreply: bool = False,
     notification: bool = False,
+    types: Optional[Tuple[int, ...]] = None,
+    anydata_fmt: str = "json",
 ) -> Optional[DNode]:
     """
     Convert a python dictionary to a DNode object given a YANG module object. The return
@@ -1254,6 +1273,34 @@ def _create_list(_parent, module, name, key_values, in_rpc_output=False):
         created.append(n[0])
         return n[0]
 
+    def _create_anydata(_parent, module, name, value, in_rpc_output=False):
+        if value is not None:
+            if isinstance(value, dict) and anydata_fmt == "json":
+                value = json.dumps(value)
+            elif not isinstance(value, str):
+                value = str(value)
+
+        n = ffi.new("struct lyd_node **")
+        flags = newval_flags(rpc_output=in_rpc_output, store_only=store_only)
+        ret = lib.lyd_new_any(
+            _parent,
+            module.cdata,
+            str2c(name),
+            str2c(value),
+            anydata_format(anydata_fmt),
+            flags,
+            n,
+        )
+        if ret != lib.LY_SUCCESS:
+            if _parent:
+                parent_path = repr(DNode.new(module.context, _parent).path())
+            else:
+                parent_path = "module %r" % module.name()
+            raise module.context.error(
+                "failed to create leaf %r as a child of %s", name, parent_path
+            )
+        created.append(n[0])
+
     schema_cache = {}
 
     def _find_schema(schema_parent, name, prefix):
@@ -1270,7 +1317,7 @@ def _find_schema(schema_parent, name, prefix):
             if schema_parent is None:
                 # there may not be any input or any output node in the rpc
                 return None, None
-        for s in schema_parent:
+        for s in schema_parent.children(types=types):
             if s.name() != name:
                 continue
             mod = s.module()
@@ -1370,6 +1417,9 @@ def _to_dnode(_dic, _schema, _parent=ffi.NULL, in_rpc_output=False):
                 n = _create_container(_parent, module, name, in_rpc_output)
                 _to_dnode(value, s, n, in_rpc_output)
 
+            elif isinstance(s, SAnydata):
+                _create_anydata(_parent, module, name, value, in_rpc_output)
+
     result = None
 
     try:

tests/test_data.py

@@ -10,6 +10,7 @@
 from _libyang import lib
 from libyang import (
     Context,
+    DAnydata,
     DAnyxml,
     DataType,
     DContainer,
@@ -21,6 +22,7 @@
     DRpc,
     IOType,
     LibyangError,
+    SNode,
 )
 from libyang.data import dict_to_dnode
 
@@ -1036,3 +1038,16 @@ def test_dnode_attrs_set_and_remove_multiple(self):
 
         attrs.remove("ietf-netconf:operation")
         self.assertEqual(len(attrs), 0)
+
+    def test_dnode_anydata_dict_to_dnode(self):
+        anydata_json = """{
+            "yolo-nodetypes:any1": {
+                "key1": "val1"
+            }
+            }"""
+        data = json.loads(anydata_json)
+        module = self.ctx.load_module("yolo-nodetypes")
+        dnode = dict_to_dnode(
+            data, module, None, validate=False, types=(SNode.ANYDATA,)
+        )
+        self.assertIsInstance(dnode, DAnydata)