# Copyright (c) 2020 6WIND S.A.
# Copyright (c) 2021 RACOM s.r.o.
# SPDX-License-Identifier: MIT

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,
    SContainer,
    SLeaf,
    SLeafList,
    SList,
    SNode,
    SNotif,
    SRpc,
    Type,
)
from .util import DataType, IOType, LibyangError, c2str, ly_array_iter, str2c


LOG = logging.getLogger(__name__)


# -------------------------------------------------------------------------------------
def implicit_flags(
    no_config: bool = False,
    no_defaults: bool = False,
    no_state: bool = False,
    output: bool = False,
) -> int:
    flags = 0
    if no_config:
        flags |= lib.LYD_IMPLICIT_NO_CONFIG
    if no_state:
        flags |= lib.LYD_IMPLICIT_NO_STATE
    if no_defaults:
        flags |= lib.LYD_IMPLICIT_NO_DEFAULTS
    if output:
        flags |= lib.LYD_IMPLICIT_OUTPUT
    return flags


# -------------------------------------------------------------------------------------
def printer_flags(
    with_siblings: bool = False,
    pretty: bool = True,
    keep_empty_containers: bool = False,
    trim_default_values: bool = False,
    include_implicit_defaults: bool = False,
) -> int:
    flags = 0
    if with_siblings:
        flags |= lib.LYD_PRINT_SIBLINGS
    if not pretty:
        flags |= lib.LYD_PRINT_SHRINK
    if keep_empty_containers:
        flags |= lib.LYD_PRINT_EMPTY_CONT
    if trim_default_values:
        flags |= lib.LYD_PRINT_WD_TRIM
    if include_implicit_defaults:
        flags |= lib.LYD_PRINT_WD_ALL
    return flags


# -------------------------------------------------------------------------------------
def data_format(fmt_string: str) -> int:
    if fmt_string == "json":
        return lib.LYD_JSON
    if fmt_string == "xml":
        return lib.LYD_XML
    if fmt_string == "lyb":
        return lib.LYD_LYB
    raise ValueError("unknown data format: %r" % fmt_string)


# -------------------------------------------------------------------------------------
def newval_flags(
    rpc_output: bool = False,
    store_only: bool = False,
    bin_value: bool = False,
    canon_value: bool = False,
    meta_clear_default: bool = False,
    update: bool = False,
    opaq: bool = False,
) -> int:
    """
    Translate from booleans to newvaloptions flags.
    """
    flags = 0
    if rpc_output:
        flags |= lib.LYD_NEW_VAL_OUTPUT
    if store_only:
        flags |= lib.LYD_NEW_VAL_STORE_ONLY
    if bin_value:
        flags |= lib.LYD_NEW_VAL_BIN
    if canon_value:
        flags |= lib.LYD_NEW_VAL_CANON
    if meta_clear_default:
        flags |= lib.LYD_NEW_META_CLEAR_DFLT
    if update:
        flags |= lib.LYD_NEW_PATH_UPDATE
    if opaq:
        flags |= lib.LYD_NEW_PATH_OPAQ
    return flags


# -------------------------------------------------------------------------------------
def parser_flags(
    no_state: bool = False,
    parse_only: bool = False,
    opaq: bool = False,
    ordered: bool = False,
    strict: bool = False,
    store_only: bool = False,
    json_null: bool = False,
    json_string_datatypes: bool = False,
) -> int:
    flags = 0
    if no_state:
        flags |= lib.LYD_PARSE_NO_STATE
    if parse_only:
        flags |= lib.LYD_PARSE_ONLY
    if opaq:
        flags |= lib.LYD_PARSE_OPAQ
    if ordered:
        flags |= lib.LYD_PARSE_ORDERED
    if strict:
        flags |= lib.LYD_PARSE_STRICT
    if store_only:
        flags |= lib.LYD_PARSE_STORE_ONLY
    if json_null:
        flags |= lib.LYD_PARSE_JSON_NULL
    if json_string_datatypes:
        flags |= lib.LYD_PARSE_JSON_STRING_DATATYPES
    return flags


# -------------------------------------------------------------------------------------
def merge_flags(
    defaults: bool = False,
    destruct: bool = False,
    with_flags: bool = False,
) -> int:
    flags = 0
    if defaults:
        flags |= lib.LYD_MERGE_DEFAULTS
    if destruct:
        flags |= lib.LYD_MERGE_DESTRUCT
    if with_flags:
        flags |= lib.LYD_MERGE_WITH_FLAGS
    return flags


# -------------------------------------------------------------------------------------
def dup_flags(
    no_meta: bool = False,
    recursive: bool = False,
    with_flags: bool = False,
    with_parents: bool = False,
) -> int:
    flags = 0
    if no_meta:
        flags |= lib.LYD_DUP_NO_META
    if recursive:
        flags |= lib.LYD_DUP_RECURSIVE
    if with_flags:
        flags |= lib.LYD_DUP_WITH_FLAGS
    if with_parents:
        flags |= lib.LYD_DUP_WITH_PARENTS
    return flags


# -------------------------------------------------------------------------------------
def data_type(dtype: DataType) -> int:
    if not isinstance(dtype, DataType):
        dtype = DataType(dtype)
    return dtype.value


# -------------------------------------------------------------------------------------
def validation_flags(
    no_state: bool = False,
    validate_present: bool = False,
    validate_multi_error: bool = False,
) -> int:
    flags = 0
    if no_state:
        flags |= lib.LYD_VALIDATE_NO_STATE
    if validate_present:
        flags |= lib.LYD_VALIDATE_PRESENT
    if validate_multi_error:
        flags |= lib.LYD_VALIDATE_MULTI_ERROR
    return flags


def diff_flags(with_defaults: bool = False) -> int:
    flags = 0
    if with_defaults:
        flags |= lib.LYD_DIFF_DEFAULTS
    return flags


# -------------------------------------------------------------------------------------
class DNodeAttrs:
    __slots__ = ("context", "parent", "cdata", "__dict__")

    def __init__(self, context: "libyang.Context", parent: "libyang.DNode"):
        self.context = context
        self.parent = parent
        self.cdata = []  # C type: "struct lyd_attr *"

    def get(self, name: str) -> Optional[str]:
        for attr_name, attr_value in self:
            if attr_name == name:
                return attr_value
        return None

    def set(self, name: str, value: str):
        attrs = ffi.new("struct lyd_attr **")
        ret = lib.lyd_new_attr(
            self.parent.cdata,
            ffi.NULL,
            str2c(name),
            str2c(value),
            attrs,
        )
        if ret != lib.LY_SUCCESS:
            raise self.context.error("cannot create attr")
        self.cdata.append(attrs[0])

    def remove(self, name: str):
        for attr in self.cdata:
            if self._get_attr_name(attr) == name:
                lib.lyd_free_attr_single(self.context.cdata, attr)
                self.cdata.remove(attr)
                break

    def __contains__(self, name: str) -> bool:
        for attr_name, _ in self:
            if attr_name == name:
                return True
        return False

    def __iter__(self) -> Iterator[Tuple[str, str]]:
        for attr in self.cdata:
            name = self._get_attr_name(attr)
            yield (name, c2str(attr.value))

    def __len__(self) -> int:
        return len(self.cdata)

    @staticmethod
    def _get_attr_name(cdata) -> str:
        if cdata.name.prefix != ffi.NULL:
            return f"{c2str(cdata.name.prefix)}:{c2str(cdata.name.name)}"
        return c2str(cdata.name.name)


# -------------------------------------------------------------------------------------
class DNode:
    """
    Data tree node.
    """

    __slots__ = ("context", "cdata", "attributes", "free_func", "__dict__")

    def __init__(self, context: "libyang.Context", cdata):
        """
        :arg context:
            The libyang.Context python object.
        :arg cdata:
            The pointer to the C structure allocated by libyang.so.
        """
        self.context = context
        self.cdata = cdata  # C type: "struct lyd_node *"
        self.attributes = None
        self.free_func = None  # type: Callable[DNode]

    def meta(self) -> Dict[str, str]:
        ret = {}
        item = self.cdata.meta
        while item != ffi.NULL:
            ret[c2str(item.name)] = c2str(
                lib.lyd_value_get_canonical(
                    self.context.cdata, ffi.addressof(item.value)
                )
            )
            item = item.next
        return ret

    def get_meta(self, name: str) -> Optional[str]:
        item = self.cdata.meta
        while item != ffi.NULL:
            if c2str(item.name) == name:
                return c2str(
                    lib.lyd_value_get_canonical(
                        self.context.cdata, ffi.addressof(item.value)
                    )
                )
            item = item.next
        return None

    def meta_free(self, name: str):
        item = self.cdata.meta
        while item != ffi.NULL:
            if c2str(item.name) == name:
                lib.lyd_free_meta_single(item)
                break
            item = item.next

    def new_meta(
        self, name: str, value: str, clear_dflt: bool = False, store_only: bool = False
    ):
        flags = newval_flags(meta_clear_default=clear_dflt, store_only=store_only)
        ret = lib.lyd_new_meta(
            ffi.NULL,
            self.cdata,
            ffi.NULL,
            str2c(name),
            str2c(value),
            flags,
            ffi.NULL,
        )
        if ret != lib.LY_SUCCESS:
            raise self.context.error("cannot create meta")

    def attrs(self) -> DNodeAttrs:
        if not self.attributes:
            self.attributes = DNodeAttrs(self.context, self)
        return self.attributes

    def add_defaults(
        self,
        no_config: bool = False,
        no_defaults: bool = False,
        no_state: bool = False,
        output: bool = False,
        only_node: bool = False,
        only_module: Optional[Module] = None,
    ):
        flags = implicit_flags(
            no_config=no_config,
            no_defaults=no_defaults,
            no_state=no_state,
            output=output,
        )
        if only_node:
            node_p = ffi.cast("struct lyd_node *", self.cdata)
            ret = lib.lyd_new_implicit_tree(node_p, flags, ffi.NULL)
        else:
            node_p = ffi.new("struct lyd_node **")
            node_p[0] = self.cdata
            if only_module is not None:
                ret = lib.lyd_new_implicit_module(
                    node_p, only_module.cdata, flags, ffi.NULL
                )
            else:
                ret = lib.lyd_new_implicit_all(
                    node_p, self.context.cdata, flags, ffi.NULL
                )

        if ret != lib.LY_SUCCESS:
            raise self.context.error("cannot get module")

    def flags(self):
        ret = {"default": False, "when_true": False, "new": False}
        if self.cdata.flags & lib.LYD_DEFAULT:
            ret["default"] = True
        if self.cdata.flags & lib.LYD_WHEN_TRUE:
            ret["when_true"] = True
        if self.cdata.flags & lib.LYD_NEW:
            ret["new"] = True
        return ret

    def is_default(self) -> bool:
        return lib.lyd_is_default(self.cdata)

    def set_when(self, value: bool):
        if value:
            self.cdata.flags |= lib.LYD_WHEN_TRUE
        else:
            self.cdata.flags &= ~lib.LYD_WHEN_TRUE

    def new_path(
        self,
        path: str,
        value: str,
        opt_update: bool = False,
        opt_output: bool = False,
        opt_opaq: bool = False,
        opt_bin_value: bool = False,
        opt_canon_value: bool = False,
        opt_store_only: bool = False,
    ):
        flags = newval_flags(
            update=opt_update,
            rpc_output=opt_output,
            opaq=opt_opaq,
            bin_value=opt_bin_value,
            canon_value=opt_canon_value,
            store_only=opt_store_only,
        )
        ret = lib.lyd_new_path(
            self.cdata, ffi.NULL, str2c(path), str2c(value), flags, ffi.NULL
        )
        if ret != lib.LY_SUCCESS:
            raise self.context.error("cannot get module")

    def insert_child(self, node):
        ret = lib.lyd_insert_child(self.cdata, node.cdata)
        if ret != lib.LY_SUCCESS:
            raise self.context.error("cannot insert node")

    def insert_sibling(self, node):
        ret = lib.lyd_insert_sibling(self.cdata, node.cdata, ffi.NULL)
        if ret != lib.LY_SUCCESS:
            raise self.context.error("cannot insert sibling")

    def insert_after(self, node):
        ret = lib.lyd_insert_after(self.cdata, node.cdata)
        if ret != lib.LY_SUCCESS:
            raise self.context.error("cannot insert sibling after")

    def insert_before(self, node):
        ret = lib.lyd_insert_before(self.cdata, node.cdata)
        if ret != lib.LY_SUCCESS:
            raise self.context.error("cannot insert sibling before")

    def name(self) -> str:
        return c2str(self.cdata.schema.name)

    def module(self) -> Module:
        if not self.cdata.schema:
            raise self.context.error("cannot get module")
        return Module(self.context, self.cdata.schema.module)

    def schema(self) -> SNode:
        return SNode.new(self.context, self.cdata.schema)

    def parent(self) -> Optional["DNode"]:
        if not self.cdata.parent:
            return None
        return self.new(self.context, self.cdata.parent)

    def next(self) -> Optional["DNode"]:
        if not self.cdata.next:
            return None
        return self.new(self.context, self.cdata.next)

    def prev(self) -> Optional["DNode"]:
        if not self.cdata.prev:
            return None
        return self.new(self.context, self.cdata.prev)

    def root(self) -> "DNode":
        node = self
        while node.parent() is not None:
            node = node.parent()
        return node

    def first_sibling(self) -> "DNode":
        n = lib.lyd_first_sibling(self.cdata)
        if n == self.cdata:
            return self
        return self.new(self.context, n)

    def siblings(self, include_self: bool = True) -> Iterator["DNode"]:
        n = lib.lyd_first_sibling(self.cdata)
        while n:
            if n == self.cdata:
                if include_self:
                    yield self
            else:
                yield self.new(self.context, n)
            n = n.next

    def find_path(self, path: str, output: bool = False):
        node = ffi.new("struct lyd_node **")
        ret = lib.lyd_find_path(self.cdata, str2c(path), output, node)
        if ret == lib.LY_SUCCESS:
            return DNode.new(self.context, node[0])
        return None

    def find_one(self, xpath: str) -> Optional["DNode"]:
        try:
            return next(self.find_all(xpath))
        except StopIteration:
            return None

    def find_all(self, xpath: str) -> Iterator["DNode"]:
        node_set = ffi.new("struct ly_set **")
        ret = lib.lyd_find_xpath(self.cdata, str2c(xpath), node_set)
        if ret != lib.LY_SUCCESS:
            raise self.context.error("cannot find path: %s", xpath)

        node_set = node_set[0]
        try:
            for i in range(node_set.count):
                n = DNode.new(self.context, node_set.dnodes[i])
                yield n
        finally:
            lib.ly_set_free(node_set, ffi.NULL)

    def eval_xpath(self, xpath: str):
        lbool = ffi.new("ly_bool *")
        ret = lib.lyd_eval_xpath(self.cdata, str2c(xpath), lbool)
        if ret != lib.LY_SUCCESS:
            raise self.context.error("cannot eva xpath: %s", xpath)
        if lbool[0]:
            return True
        return False

    def path(self) -> str:
        return self._get_path(self.cdata)

    def validate(
        self,
        no_state: bool = False,
        validate_present: bool = False,
        rpc: bool = False,
        rpcreply: bool = False,
        notification: bool = False,
        dep_tree: Optional["DNode"] = None,
    ) -> None:
        dtype = None
        if rpc:
            dtype = DataType.RPC_YANG
        elif rpcreply:
            dtype = DataType.REPLY_YANG
        elif notification:
            dtype = DataType.NOTIF_YANG

        if dtype is None:
            self.validate_all(no_state, validate_present)
        else:
            self.validate_op(dtype, dep_tree)

    def validate_all(
        self,
        no_state: bool = False,
        validate_present: bool = False,
    ) -> None:
        if self.cdata.parent:
            raise self.context.error("validation is only supported on top-level nodes")
        flags = validation_flags(
            no_state=no_state,
            validate_present=validate_present,
        )
        node_p = ffi.new("struct lyd_node **")
        node_p[0] = self.cdata
        ret = lib.lyd_validate_all(node_p, self.context.cdata, flags, ffi.NULL)
        if ret != lib.LY_SUCCESS:
            raise self.context.error("validation failed")

    def validate_op(
        self,
        dtype: DataType,
        dep_tree: Optional["DNode"] = None,
    ) -> None:
        dtype = data_type(dtype)
        ret = lib.lyd_validate_op(
            self.cdata,
            ffi.NULL if dep_tree is None else dep_tree.cdata,
            dtype,
            ffi.NULL,
        )
        if ret != lib.LY_SUCCESS:
            raise self.context.error("validation failed")

    def diff(
        self,
        other: "DNode",
        no_siblings: bool = False,
        with_defaults: bool = False,
    ) -> "DNode":
        flags = diff_flags(with_defaults=with_defaults)
        node_p = ffi.new("struct lyd_node **")
        if no_siblings:
            ret = lib.lyd_diff_tree(self.cdata, other.cdata, flags, node_p)
        else:
            ret = lib.lyd_diff_siblings(self.cdata, other.cdata, flags, node_p)

        if ret != lib.LY_SUCCESS:
            raise self.context.error("diff error")

        if node_p[0] == ffi.NULL:
            return None

        return self.new(self.context, node_p[0])

    def diff_apply(self, diff_node: "DNode") -> None:
        node_p = ffi.new("struct lyd_node **")
        node_p[0] = self.cdata

        ret = lib.lyd_diff_apply_all(node_p, diff_node.cdata)
        if ret != lib.LY_SUCCESS:
            raise self.context.error("apply diff error")

    def duplicate(
        self,
        with_siblings: bool = False,
        no_meta: bool = False,
        recursive: bool = False,
        with_flags: bool = False,
        with_parents: bool = False,
        parent: Optional["DNode"] = None,
    ) -> "DNode":
        flags = dup_flags(
            no_meta=no_meta,
            recursive=recursive,
            with_flags=with_flags,
            with_parents=with_parents,
        )

        if parent is not None:
            parent = parent.cdata
        else:
            parent = ffi.NULL

        node = ffi.new("struct lyd_node **")
        if with_siblings:
            lib.lyd_dup_siblings(self.cdata, parent, flags, node)
        else:
            lib.lyd_dup_single(self.cdata, parent, flags, node)

        return DNode.new(self.context, node[0])

    def merge_module(
        self,
        source: "DNode",
        defaults: bool = False,
        destruct: bool = False,
        with_flags: bool = False,
    ) -> None:
        flags = merge_flags(defaults=defaults, destruct=destruct, with_flags=with_flags)
        node_p = ffi.new("struct lyd_node **")
        node_p[0] = self.cdata
        ret = lib.lyd_merge_module(
            node_p, source.cdata, ffi.NULL, ffi.NULL, ffi.NULL, flags
        )
        if ret != lib.LY_SUCCESS:
            raise self.context.error("merge failed")

    def merge(
        self,
        source: "DNode",
        with_siblings: bool = False,
        defaults: bool = False,
        destruct: bool = False,
        with_flags: bool = False,
    ) -> None:
        flags = merge_flags(defaults=defaults, destruct=destruct, with_flags=with_flags)
        node_p = ffi.new("struct lyd_node **")
        node_p[0] = self.cdata
        if with_siblings:
            ret = lib.lyd_merge_siblings(node_p, source.cdata, flags)
        else:
            ret = lib.lyd_merge_tree(node_p, source.cdata, flags)

        if ret != lib.LY_SUCCESS:
            raise self.context.error("merge failed")

        self.cdata = node_p[0]

    def iter_tree(self) -> Iterator["DNode"]:
        n = next_n = self.cdata
        while n != ffi.NULL:
            yield self.new(self.context, n)

            next_n = lib.lyd_child(n)
            if next_n == ffi.NULL:
                if n == self.cdata:
                    break
                next_n = n.next
            while next_n == ffi.NULL:
                n = n.parent
                if n.parent == self.cdata.parent:
                    break
                next_n = n.next
            n = next_n

    def print(
        self,
        fmt: str,
        out_type: IOType,
        out_target: Union[IO, str, None] = None,
        with_siblings: bool = False,
        pretty: bool = True,
        keep_empty_containers: bool = False,
        trim_default_values: bool = False,
        include_implicit_defaults: bool = False,
    ) -> Union[str, bytes]:
        flags = printer_flags(
            pretty=pretty,
            keep_empty_containers=keep_empty_containers,
            trim_default_values=trim_default_values,
            include_implicit_defaults=include_implicit_defaults,
        )
        fmt = data_format(fmt)
        out_data = ffi.new("struct ly_out **")

        if out_type == IOType.FD:
            raise NotImplementedError

        if out_type == IOType.FILE:
            raise NotImplementedError

        if out_type == IOType.FILEPATH:
            raise NotImplementedError

        if out_type != IOType.MEMORY:
            raise ValueError("no input specified")

        buf = ffi.new("char **")
        ret = lib.ly_out_new_memory(buf, 0, out_data)
        if ret != lib.LY_SUCCESS:
            raise self.context.error("failed to initialize output target")

        if with_siblings:
            ret = lib.lyd_print_all(out_data[0], self.cdata, fmt, flags)
        else:
            ret = lib.lyd_print_tree(out_data[0], self.cdata, fmt, flags)

        lib.ly_out_free(out_data[0], ffi.NULL, 0)
        if ret != lib.LY_SUCCESS:
            raise self.context.error("failed to write data")

        return c2str(buf[0], decode=True)

    def print_mem(
        self,
        fmt: str,
        with_siblings: bool = False,
        pretty: bool = True,
        keep_empty_containers: bool = False,
        trim_default_values: bool = False,
        include_implicit_defaults: bool = False,
    ) -> Union[str, bytes]:
        flags = printer_flags(
            with_siblings=with_siblings,
            pretty=pretty,
            keep_empty_containers=keep_empty_containers,
            trim_default_values=trim_default_values,
            include_implicit_defaults=include_implicit_defaults,
        )
        buf = ffi.new("char **")
        fmt = data_format(fmt)
        ret = lib.lyd_print_mem(buf, self.cdata, fmt, flags)
        if ret != lib.LY_SUCCESS:
            raise self.context.error("cannot print node")
        try:
            if fmt == lib.LYD_LYB:
                # binary format, do not convert to unicode
                return c2str(buf[0], decode=False)
            return c2str(buf[0], decode=True)
        finally:
            lib.free(buf[0])

    def print_file(
        self,
        fileobj: IO,
        fmt: str,
        with_siblings: bool = False,
        pretty: bool = True,
        keep_empty_containers: bool = False,
        trim_default_values: bool = False,
        include_implicit_defaults: bool = False,
    ) -> None:
        flags = printer_flags(
            with_siblings=with_siblings,
            pretty=pretty,
            keep_empty_containers=keep_empty_containers,
            trim_default_values=trim_default_values,
            include_implicit_defaults=include_implicit_defaults,
        )
        fmt = data_format(fmt)
        out = ffi.new("struct ly_out **")
        ret = lib.ly_out_new_fd(fileobj.fileno(), out)
        if ret != lib.LY_SUCCESS:
            raise self.context.error("cannot allocate output data")
        if with_siblings:
            ret = lib.lyd_print_all(out[0], self.cdata, fmt, flags)
        else:
            ret = lib.lyd_print_tree(out[0], self.cdata, fmt, flags)
        if ret != lib.LY_SUCCESS:
            raise self.context.error("cannot print node")

    def should_print(
        self,
        include_implicit_defaults: bool = False,
        trim_default_values: bool = False,
        keep_empty_containers: bool = False,
    ) -> bool:
        """
        Check if a node should be "printed".

        :arg bool include_implicit_defaults:
            Include implicit default nodes.
        :arg bool trim_default_values:
            Exclude nodes with the value equal to their default value.
        :arg bool keep_empty_containers:
            Preserve empty non-presence containers.
        """
        flags = printer_flags(
            include_implicit_defaults=include_implicit_defaults,
            trim_default_values=trim_default_values,
            keep_empty_containers=keep_empty_containers,
        )
        return bool(lib.lyd_node_should_print(self.cdata, flags))

    def print_dict(
        self,
        strip_prefixes: bool = True,
        absolute: bool = True,
        with_siblings: bool = False,
        include_implicit_defaults: bool = False,
        trim_default_values: bool = False,
        keep_empty_containers: bool = False,
    ) -> Dict[str, Any]:
        """
        Convert a DNode object to a python dictionary.

        The format is inspired by the YANG JSON format described in :rfc:`7951` but has
        some differences:

        * ``int64`` and ``uint64`` values are represented by python ``int`` values
          instead of string values.
        * ``decimal64`` values are represented by python ``float`` values instead of
          string values.
        * ``empty`` values are represented by python ``None`` values instead of
          ``[None]`` list instances. To check if an ``empty`` leaf is set in a
          container, you should use the idiomatic ``if "foo" in container:`` construct.

        :arg bool strip_prefixes:
            If True (the default), module prefixes are stripped from dictionary keys. If
            False, dictionary keys are in the form ``<module>:<name>``.
        :arg bool absolute:
            If True (the default), always return a dictionary containing the complete
            tree starting from the root.
        :arg bool with_siblings:
            If True, include the node's siblings.
        :arg bool include_implicit_defaults:
            Include implicit default nodes.
        :arg bool trim_default_values:
            Exclude nodes with the value equal to their default value.
        :arg bool keep_empty_containers:
            Preserve empty non-presence containers.
        """
        flags = printer_flags(
            include_implicit_defaults=include_implicit_defaults,
            trim_default_values=trim_default_values,
            keep_empty_containers=keep_empty_containers,
        )

        name_cache = {}

        def _node_name(node):
            name = name_cache.get(node.schema)
            if name is None:
                if strip_prefixes:
                    name = c2str(node.schema.name)
                else:
                    mod = node.schema.module
                    name = "%s:%s" % (c2str(mod.name), c2str(node.schema.name))
                name_cache[node.schema] = name
            return name

        list_keys_cache = {}

        def _init_yang_list(snode):
            if snode.flags & lib.LYS_ORDBY_USER:
                return []  # ordered list, return an empty builtin list

            # unordered lists
            if snode.nodetype == SNode.LEAFLIST:
                return KeyedList(key_name=None)

            if snode not in list_keys_cache:
                keys = []
                child = lib.lysc_node_child(snode)
                while child:
                    if child.flags & lib.LYS_KEY:
                        if strip_prefixes:
                            keys.append(c2str(child.name))
                        else:
                            keys.append(
                                "%s:%s" % (c2str(child.module.name), c2str(child.name))
                            )
                    child = child.next
                if len(keys) == 1:
                    list_keys_cache[snode] = keys[0]
                else:
                    list_keys_cache[snode] = tuple(keys)

            return KeyedList(key_name=list_keys_cache[snode])

        def _to_dict(node, parent_dic):
            if not lib.lyd_node_should_print(node, flags):
                return
            name = _node_name(node)
            if node.schema.nodetype == SNode.LIST:
                list_element = {}

                child = lib.lyd_child(node)
                while child:
                    _to_dict(child, list_element)
                    child = child.next
                if name not in parent_dic:
                    parent_dic[name] = _init_yang_list(node.schema)
                parent_dic[name].append(list_element)
            elif node.schema.nodetype & (
                SNode.CONTAINER | SNode.RPC | SNode.ACTION | SNode.NOTIF
            ):
                container = {}
                child = lib.lyd_child(node)
                while child:
                    _to_dict(child, container)
                    child = child.next
                parent_dic[name] = container
            elif node.schema.nodetype == SNode.LEAFLIST:
                if name not in parent_dic:
                    parent_dic[name] = _init_yang_list(node.schema)
                parent_dic[name].append(DLeaf.cdata_leaf_value(node, self.context))
            elif node.schema.nodetype == SNode.LEAF:
                parent_dic[name] = DLeaf.cdata_leaf_value(node, self.context)

        dic = {}
        dnode = self
        if absolute:
            dnode = dnode.root()
        if with_siblings:
            for sib in dnode.siblings():
                _to_dict(sib.cdata, dic)
        else:
            _to_dict(dnode.cdata, dic)
        return dic

    def merge_data_dict(
        self,
        dic: Dict[str, Any],
        no_state: bool = False,
        validate_present: bool = False,
        validate: bool = True,
        strict: bool = False,
        rpc: bool = False,
        rpcreply: bool = False,
    ) -> Optional["DNode"]:
        """
        Merge a python dictionary into this node. The returned value is the first
        created node.

        :arg dic:
            The python dictionary to convert.
        :arg no_state:
            Consider state data not allowed and raise an error during validation if they are found.
        :arg validate_present:
            Validate result of the operation against schema.
        :arg validate:
            Run validation on result of the operation.
        :arg strict:
            Instead of ignoring data without schema definition, raise an error.
        :arg rpc:
            Data represents RPC or action input parameters.
        :arg rpcreply:
            Data represents RPC or action output parameters.
        """
        return dict_to_dnode(
            dic,
            self.module(),
            parent=self,
            no_state=no_state,
            validate_present=validate_present,
            validate=validate,
            strict=strict,
            rpc=rpc,
            rpcreply=rpcreply,
        )

    def unlink(self, with_siblings: bool = False) -> None:
        if with_siblings:
            lib.lyd_unlink_siblings(self.cdata)
        else:
            lib.lyd_unlink_tree(self.cdata)

    def free_internal(self, with_siblings: bool = True) -> None:
        if with_siblings:
            lib.lyd_free_all(self.cdata)
        else:
            lib.lyd_free_tree(self.cdata)

    def free(self, with_siblings: bool = True) -> None:
        try:
            if self.free_func:
                self.free_func(self)  # pylint: disable=not-callable
            else:
                self.free_internal(with_siblings)
        finally:
            self.cdata = ffi.NULL

    def leafref_link_node_tree(self) -> None:
        """
        Traverse through data tree including root node siblings and adds
        leafrefs links to the given nodes.

        Requires leafref_linking to be set on the libyang context.
        """
        lib.lyd_leafref_link_node_tree(self.cdata)

    def leafref_nodes(self) -> Iterator["DNode"]:
        """
        Gets the nodes that are referring to this node.

        Requires leafref_linking to be set on the libyang context.
        """
        term_node = ffi.cast("struct lyd_node_term *", self.cdata)
        out = ffi.new("const struct lyd_leafref_links_rec **")
        if lib.lyd_leafref_get_links(term_node, out) != lib.LY_SUCCESS:
            return
        for n in ly_array_iter(out[0].leafref_nodes):
            yield DNode.new(self.context, n)

    def target_nodes(self) -> Iterator["DNode"]:
        """
        Gets the target nodes that are referred by this node.

        Requires leafref_linking to be set on the libyang context.
        """
        term_node = ffi.cast("struct lyd_node_term *", self.cdata)
        out = ffi.new("const struct lyd_leafref_links_rec **")
        if lib.lyd_leafref_get_links(term_node, out) != lib.LY_SUCCESS:
            return
        for n in ly_array_iter(out[0].target_nodes):
            yield DNode.new(self.context, n)

    def __repr__(self):
        cls = self.__class__
        return "<%s.%s: %s>" % (cls.__module__, cls.__name__, str(self))

    def __str__(self):
        return self.name()

    NODETYPE_CLASS = {}

    @classmethod
    def register(cls, *nodetypes):
        def _decorator(nodeclass):
            for t in nodetypes:
                cls.NODETYPE_CLASS[t] = nodeclass
            return nodeclass

        return _decorator

    @classmethod
    def new(cls, context: "libyang.Context", cdata) -> "DNode":
        cdata = ffi.cast("struct lyd_node *", cdata)
        if not cdata.schema:
            schemas = list(context.find_path(cls._get_path(cdata)))
            if len(schemas) != 1:
                raise LibyangError("Unable to determine schema")
            nodecls = cls.NODETYPE_CLASS.get(schemas[0].nodetype(), None)
        else:
            nodecls = cls.NODETYPE_CLASS.get(cdata.schema.nodetype, None)
        if nodecls is None:
            raise TypeError("node type %s not implemented" % cdata.schema.nodetype)
        return nodecls(context, cdata)

    @staticmethod
    def _get_path(cdata) -> str:
        path = lib.lyd_path(cdata, lib.LYD_PATH_STD, ffi.NULL, 0)
        try:
            return c2str(path)
        finally:
            lib.free(path)


# -------------------------------------------------------------------------------------
@DNode.register(SNode.CONTAINER)
class DContainer(DNode):
    def create_path(
        self,
        path: str,
        value: Any = None,
        rpc_output: bool = False,
        store_only: bool = False,
    ) -> Optional[DNode]:
        return self.context.create_data_path(
            path, parent=self, value=value, rpc_output=rpc_output, store_only=store_only
        )

    def children(self, no_keys=False) -> Iterator[DNode]:
        if no_keys:
            child = lib.lyd_child_no_keys(self.cdata)
        else:
            child = lib.lyd_child(self.cdata)

        while child:
            if child.schema != ffi.NULL:
                yield DNode.new(self.context, child)
            child = child.next

    def __iter__(self):
        return self.children()


# -------------------------------------------------------------------------------------
@DNode.register(SNode.RPC)
@DNode.register(SNode.ACTION)
class DRpc(DContainer):
    pass


# -------------------------------------------------------------------------------------
@DNode.register(SNode.LIST)
class DList(DContainer):
    pass


# -------------------------------------------------------------------------------------
@DNode.register(SNode.LEAF)
class DLeaf(DNode):
    def value(self) -> Any:
        return DLeaf.cdata_leaf_value(self.cdata, self.context)

    @staticmethod
    def cdata_leaf_value(cdata, context: "libyang.Context" = None) -> Any:
        val = lib.lyd_get_value(cdata)
        if val == ffi.NULL:
            return None

        val = c2str(val)
        if cdata.schema == ffi.NULL:
            # opaq node
            return val

        node_term = ffi.cast("struct lyd_node_term *", cdata)

        # inspired from libyang lyd_value_validate
        val_type = Type(context, node_term.value.realtype, None).base()
        if val_type == Type.UNION:
            val_type = Type(
                context, node_term.value.subvalue.value.realtype, None
            ).base()

        if val_type in Type.STR_TYPES:
            return val
        if val_type in Type.NUM_TYPES:
            return int(val)
        if val_type == Type.BOOL:
            return val == "true"
        if val_type == Type.DEC64:
            return float(val)
        if val_type == Type.EMPTY:
            return None
        return val


# -------------------------------------------------------------------------------------
@DNode.register(SNode.LEAFLIST)
class DLeafList(DLeaf):
    pass


# -------------------------------------------------------------------------------------
@DNode.register(SNode.NOTIF)
class DNotif(DContainer):
    pass


# -------------------------------------------------------------------------------------
@DNode.register(SNode.ANYXML)
class DAnyxml(DNode):
    def value(self, fmt: str = "xml"):
        anystr = ffi.new("char **", ffi.NULL)
        ret = lib.lyd_any_value_str(self.cdata, anystr)
        if ret != lib.LY_SUCCESS:
            raise self.context.error("cannot get data")
        return c2str(anystr[0])


# -------------------------------------------------------------------------------------
@DNode.register(SNode.ANYDATA)
class DAnydata(DNode):
    def value(self, fmt: str = "xml"):
        anystr = ffi.new("char **", ffi.NULL)
        ret = lib.lyd_any_value_str(self.cdata, anystr)
        if ret != lib.LY_SUCCESS:
            raise self.context.error("cannot get data")
        return c2str(anystr[0])


# -------------------------------------------------------------------------------------
def dict_to_dnode(
    dic: Dict[str, Any],
    module: Module,
    parent: Optional[DNode] = None,
    no_state: bool = False,
    validate_present: bool = False,
    validate: bool = True,
    strict: bool = False,
    rpc: bool = False,
    rpcreply: bool = False,
    notification: bool = False,
    store_only: bool = False,
) -> Optional[DNode]:
    """
    Convert a python dictionary to a DNode object given a YANG module object. The return
    value is the first created node. If parent is not set, a top-level node is returned.

    :arg dic:
        The python dictionary to convert.
    :arg module:
        The libyang Module object associated with the dictionary.
    :arg parent:
        Optional parent to update. If not specified a new top-level DNode will be
        created.
    :arg no_state:
        Consider state data not allowed and raise an error during validation if they are found.
    :arg validate_present:
        Validate result of the operation against schema.
    :arg validate:
        Run validation on result of the operation.
    :arg strict:
        Instead of ignoring data without schema definition, raise an error.
    :arg rpc:
        Data represents RPC or action input parameters.
    :arg rpcreply:
        Data represents RPC or action output parameters.
    :arg notification:
        Data represents notification parameters.
    :arg store_only:
        Data are being stored regardless of type validation (length, range, pattern, etc.)
    """
    if not dic:
        return None

    if not isinstance(dic, dict):
        raise TypeError("dic argument must be a python dict")
    if not isinstance(module, Module):
        raise TypeError("module argument must be a Module object")
    if parent is not None and not isinstance(parent, DNode):
        raise TypeError("parent argument must be a DNode object or None")

    created = []

    def _create_leaf(_parent, module, name, value, in_rpc_output=False):
        if value is not None:
            if isinstance(value, bool):
                value = str(value).lower()
            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_term(
            _parent,
            module.cdata,
            str2c(name),
            str2c(value),
            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])

    def _create_container(_parent, module, name, in_rpc_output=False):
        n = ffi.new("struct lyd_node **")
        ret = lib.lyd_new_inner(_parent, module.cdata, str2c(name), in_rpc_output, 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 container/list/rpc %r as a child of %s",
                name,
                parent_path,
            )
        created.append(n[0])
        return n[0]

    def _create_list(_parent, module, name, key_values, in_rpc_output=False):
        n = ffi.new("struct lyd_node **")
        flags = newval_flags(rpc_output=in_rpc_output, store_only=store_only)
        ret = lib.lyd_new_list(
            _parent,
            module.cdata,
            str2c(name),
            flags,
            n,
            *[str2c(str(i)) for i in key_values],
        )
        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 container/list/rpc %r as a child of %s",
                name,
                parent_path,
            )
        created.append(n[0])
        return n[0]

    schema_cache = {}

    def _find_schema(schema_parent, name, prefix):
        cache_key = (schema_parent.cdata, name, prefix)
        snode, module = schema_cache.get(cache_key, (None, None))
        if snode is not None:
            return snode, module
        if isinstance(schema_parent, SRpc):
            if rpcreply:
                schema_parent = schema_parent.output()
            else:
                schema_parent = schema_parent.input()

            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:
            if s.name() != name:
                continue
            mod = s.module()
            if prefix is not None and mod.name() != prefix:
                continue
            snode = s
            module = mod
            break
        schema_cache[cache_key] = (snode, module)
        return snode, module

    keys_cache = {}

    def _dic_keys(_dic, _schema):
        if isinstance(_schema, SList):
            # list keys must be first and in the order specified in the schema
            list_keys = keys_cache.get(_schema.cdata, None)
            if list_keys is None:
                list_keys = tuple(k.name() for k in _schema.keys())
                keys_cache[_schema.cdata] = list_keys
            keys = []
            for k in list_keys:
                if k in _dic:
                    keys.append(k)
            keys.extend(_dic.keys() - list_keys)
            return keys
        return _dic.keys()

    def _to_dnode(_dic, _schema, _parent=ffi.NULL, in_rpc_output=False):
        for key in _dic_keys(_dic, _schema):
            if ":" in key:
                prefix, name = key.split(":")
            else:
                prefix, name = None, key

            s, module = _find_schema(_schema, name, prefix)
            if not s:
                if isinstance(_schema, Module):
                    path = _schema.name()
                elif isinstance(_schema, SNode):
                    path = _schema.schema_path()
                else:
                    path = str(_schema)
                if strict:
                    raise LibyangError("%s: unknown element %r" % (path, key))
                LOG.warning("%s: skipping unknown element %r", path, key)
                continue

            value = _dic[key]

            if isinstance(s, SLeaf):
                _create_leaf(_parent, module, name, value, in_rpc_output)

            elif isinstance(s, SLeafList):
                if not isinstance(value, (list, tuple)):
                    raise TypeError(
                        "%s: python value is not a list/tuple: %r"
                        % (s.schema_path(), value)
                    )
                for v in value:
                    _create_leaf(_parent, module, name, v, in_rpc_output)

            elif isinstance(s, SRpc):
                n = _create_container(_parent, module, name, in_rpc_output)
                _to_dnode(value, s, n, rpcreply)

            elif isinstance(s, SContainer):
                n = _create_container(_parent, module, name, in_rpc_output)
                _to_dnode(value, s, n, in_rpc_output)

            elif isinstance(s, SList):
                if not isinstance(value, (list, tuple)):
                    raise TypeError(
                        "%s: python value is not a list/tuple: %r"
                        % (s.schema_path(), value)
                    )

                keys = [i.name() for i in s.keys()]
                for v in value:
                    if not isinstance(v, dict):
                        raise TypeError(
                            "%s: list element is not a dict: %r"
                            % (_schema.schema_path(), v)
                        )
                    val = v.copy()
                    key_values = []
                    for k in keys:
                        try:
                            key_values.append(val.pop(k))
                        except KeyError as e:
                            raise ValueError("Missing key %s in the list" % (k)) from e

                    n = _create_list(_parent, module, name, key_values, in_rpc_output)
                    _to_dnode(val, s, n, in_rpc_output)

            elif isinstance(s, SNotif):
                n = _create_container(_parent, module, name, in_rpc_output)
                _to_dnode(value, s, n, in_rpc_output)

    result = None

    try:
        if parent is not None:
            _parent = parent.cdata
            _schema_parent = parent.schema()
        else:
            _parent = ffi.NULL
            _schema_parent = module

        _to_dnode(
            dic,
            _schema_parent,
            _parent,
            in_rpc_output=rpcreply and isinstance(parent, DRpc),
        )
        if created:
            result = DNode.new(module.context, created[0])
            if validate:
                result.root().validate(
                    no_state=no_state,
                    validate_present=validate_present,
                    rpc=rpc,
                    rpcreply=rpcreply,
                    notification=notification,
                )

    except:
        for c in reversed(created):
            lib.lyd_free_tree(c)
        raise

    return result