diff --git a/clixon/clixon.py b/clixon/clixon.py
index 25f7728..1f1e7e9 100644
--- a/clixon/clixon.py
+++ b/clixon/clixon.py
@@ -178,24 +178,37 @@ def commit(self) -> None:
if self.__push:
self.push()
- def get_root(self, path: Optional[str] = None) -> object:
+ def get_root(
+ self,
+ path: Optional[str] = None,
+ xpath: Optional[str] = "/",
+ namespaces: Optional[dict] = None,
+ ) -> object:
"""
- Return the root object or a specific element at the given path.
+ Return the root object or a specific element, with optional server-side filtering.
Examples:
root = clixon.get_root() # Returns entire root
- device = clixon.get_root("devices/device[0]") # Returns first device
- config = clixon.get_root("devices/device[name='r1']/config") # Returns config for device 'r1'
+ device = clixon.get_root(path="devices/device[0]") # Returns first device (client-side navigation)
+ config = clixon.get_root(path="devices/device[name='r1']/config") # Returns config for device 'r1'
+ services = clixon.get_root(xpath="/services") # Returns only services subtree (server-side filter)
+ l2c = clixon.get_root(xpath="/services/l2c:l2c", namespaces={"l2c": "http://example.com/l2c"}) # Filtered with namespace
- :param path: Optional path to a specific element (e.g., "devices/device[0]"). If None, returns entire root.
+ :param path: Optional path to a specific element (e.g., "devices/device[0]"). Applied client-side after retrieval.
:type path: Optional[str]
+ :param xpath: XPath expression to filter the config server-side (default '/')
+ :type xpath: Optional[str]
+ :param namespaces: Dict of namespace prefixes to URIs for xpath (optional)
+ :type namespaces: Optional[dict]
:return: Root object (if path is None) or element at path (if path is provided). Returns None if path is invalid.
:rtype: object
"""
logger.debug("Updating root object")
- config = rpc_config_get(user=self.__user, source=self.__source)
+ config = rpc_config_get(
+ user=self.__user, source=self.__source, xpath=xpath, namespaces=namespaces
+ )
send(self.__socket, config, pp)
data = read(self.__socket, pp)
diff --git a/clixon/netconf.py b/clixon/netconf.py
index b4844b1..9fc1737 100644
--- a/clixon/netconf.py
+++ b/clixon/netconf.py
@@ -25,6 +25,11 @@ class RPCTypes(Enum):
CONTROLLER_NS = {"xmlns": "http://clicon.org/controller"}
+CONTROLLER_NS_PREFIX = "clixon-controller"
+CONTROLLER_NS_URI = "http://clicon.org/controller"
+# Top-level elements in clixon-controller namespace
+CONTROLLER_ELEMENTS = ["services", "devices"]
+
BASE_ATTRIBUTES = {
"xmlns": "urn:ietf:params:xml:ns:netconf:base:1.0",
"message-id": "42",
@@ -33,21 +38,44 @@ class RPCTypes(Enum):
def rpc_config_get(
- user: Optional[str] = None, source: Optional[str] = "actions"
-) -> Element:
+ user: Optional[str] = None,
+ source: Optional[str] = "actions",
+ xpath: Optional[str] = "/",
+ namespaces: Optional[dict] = None,
+):
"""
- Create a get-config RPC element.
-
- :param user: User name
- :type user: str
- :param source: Source of the configuration
- :type source: str
- :return: RPC element
- :rtype: Element
+ Create a get-config RPC element with optional xpath filter.
+ :param user: Username (optional)
+ :param source: Source of config ('actions' by default)
+ :param xpath: XPath string to filter config (default '/')
+ :param namespaces: Dict of namespace prefixes to URIs for xpath (optional)
+ :return: RPC element for NETCONF get-config
"""
attributes = {}
- xpath_attributes = {"nc:type": "xpath", "nc:select": "/"}
+
+ # Auto-prepend clixon-controller namespace for known top-level elements
+ for elem in CONTROLLER_ELEMENTS:
+ # Match /services or /services/... but not /clixon-controller:services
+ if xpath.startswith(f"/{elem}") and not xpath.startswith(
+ f"/{CONTROLLER_NS_PREFIX}:"
+ ):
+ xpath = xpath.replace(f"/{elem}", f"/{CONTROLLER_NS_PREFIX}:{elem}", 1)
+ break
+
+ xpath_attributes = {"nc:type": "xpath", "nc:select": xpath}
+
+ # Default namespaces for xpath - always include clixon-controller
+ default_namespaces = {
+ CONTROLLER_NS_PREFIX: CONTROLLER_NS_URI,
+ }
+
+ # Merge default namespaces with user-provided ones (user takes precedence)
+ all_namespaces = {**default_namespaces, **(namespaces or {})}
+
+ # Add namespace declarations for xpath prefixes
+ for prefix, uri in all_namespaces.items():
+ xpath_attributes[f"xmlns:{prefix}"] = uri
if not user:
user = getpass.getuser()
diff --git a/tests/test_netconf.py b/tests/test_netconf.py
index c6c296f..a5c5122 100644
--- a/tests/test_netconf.py
+++ b/tests/test_netconf.py
@@ -42,7 +42,7 @@ def test_rpc_config_get():
Test the rpc_config_get function.
"""
- xmlstr = f""""""
+ xmlstr = f""""""
root = netconf.rpc_config_get()
@@ -54,7 +54,7 @@ def test_rpc_config_get_user():
Test the rpc_config_get function with user.
"""
- xmlstr = f""""""
+ xmlstr = f""""""
root = netconf.rpc_config_get(user="nisse")
@@ -456,3 +456,58 @@ def test_rpc_devices_get():
xmlstr0 = f"""explicit"""
assert netconf.rpc_devices_get().dumps() == xmlstr0
+
+
+def test_rpc_config_get_with_xpath():
+ """
+ Test the rpc_config_get function with custom xpath.
+ """
+
+ xmlstr = f""""""
+
+ root = netconf.rpc_config_get(xpath="/services")
+
+ assert root.dumps() == xmlstr
+
+
+def test_rpc_config_get_with_xpath_and_namespaces():
+ """
+ Test the rpc_config_get function with custom xpath and namespaces.
+ """
+
+ xmlstr = f""""""
+
+ namespaces = {"l2c": "http://example.com/l2c"}
+ root = netconf.rpc_config_get(xpath="/services/l2c:l2c", namespaces=namespaces)
+
+ assert root.dumps() == xmlstr
+
+
+def test_rpc_config_get_with_xpath_different_source():
+ """
+ Test the rpc_config_get function with xpath and different source.
+ """
+
+ xmlstr = f""""""
+
+ root = netconf.rpc_config_get(source="candidate", xpath="/devices")
+
+ assert root.dumps() == xmlstr
+
+
+def test_rpc_config_get_with_multiple_namespaces():
+ """
+ Test the rpc_config_get function with multiple custom namespaces.
+ """
+
+ xmlstr = f""""""
+
+ namespaces = {
+ "l2c": "http://example.com/l2c",
+ "custom": "http://example.com/custom",
+ }
+ root = netconf.rpc_config_get(
+ xpath="/services/l2c:l2c[l2c:service-name='test']", namespaces=namespaces
+ )
+
+ assert root.dumps() == xmlstr