Add get_root_filtered that allows user to specify xpath

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)

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()

tests/test_netconf.py

@@ -42,7 +42,7 @@ def test_rpc_config_get():
     Test the rpc_config_get function.
     """
 
-    xmlstr = f"""<rpc xmlns="urn:ietf:params:xml:ns:netconf:base:1.0" username="{user}" xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0" message-id="42"><get-config><source><actions xmlns="http://clicon.org/controller"/></source><nc:filter nc:type="xpath" nc:select="/"/></get-config></rpc>"""
+    xmlstr = f"""<rpc xmlns="urn:ietf:params:xml:ns:netconf:base:1.0" username="{user}" xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0" message-id="42"><get-config><source><actions xmlns="http://clicon.org/controller"/></source><nc:filter nc:type="xpath" nc:select="/" xmlns:clixon-controller="http://clicon.org/controller"/></get-config></rpc>"""
 
     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"""<rpc xmlns="urn:ietf:params:xml:ns:netconf:base:1.0" username="nisse" xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0" message-id="42"><get-config><source><actions xmlns="http://clicon.org/controller"/></source><nc:filter nc:type="xpath" nc:select="/"/></get-config></rpc>"""
+    xmlstr = f"""<rpc xmlns="urn:ietf:params:xml:ns:netconf:base:1.0" username="nisse" xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0" message-id="42"><get-config><source><actions xmlns="http://clicon.org/controller"/></source><nc:filter nc:type="xpath" nc:select="/" xmlns:clixon-controller="http://clicon.org/controller"/></get-config></rpc>"""
 
     root = netconf.rpc_config_get(user="nisse")
 
@@ -456,3 +456,58 @@ def test_rpc_devices_get():
     xmlstr0 = f"""<rpc xmlns="urn:ietf:params:xml:ns:netconf:base:1.0" xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0" cl:username="{user}" xmlns:cl="http://clicon.org/lib" message-id="42"><get cl:content="all" xmlns:cl="http://clicon.org/lib"><nc:filter nc:type="xpath" nc:select="co:devices/co:device/co:name | co:devices/co:device/co:conn-state | co:devices/co:device/co:conn-state-timestamp | co:devices/co:device/co:logmsg" xmlns:co="http://clicon.org/controller"/><with-defaults xmlns="urn:ietf:params:xml:ns:yang:ietf-netconf-with-defaults">explicit</with-defaults></get></rpc>"""
 
     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"""<rpc xmlns="urn:ietf:params:xml:ns:netconf:base:1.0" username="{user}" xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0" message-id="42"><get-config><source><actions xmlns="http://clicon.org/controller"/></source><nc:filter nc:type="xpath" nc:select="/clixon-controller:services" xmlns:clixon-controller="http://clicon.org/controller"/></get-config></rpc>"""
+
+    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"""<rpc xmlns="urn:ietf:params:xml:ns:netconf:base:1.0" username="{user}" xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0" message-id="42"><get-config><source><actions xmlns="http://clicon.org/controller"/></source><nc:filter nc:type="xpath" nc:select="/clixon-controller:services/l2c:l2c" xmlns:clixon-controller="http://clicon.org/controller" xmlns:l2c="http://example.com/l2c"/></get-config></rpc>"""
+
+    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"""<rpc xmlns="urn:ietf:params:xml:ns:netconf:base:1.0" username="{user}" xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0" message-id="42"><get-config><source><candidate/></source><nc:filter nc:type="xpath" nc:select="/clixon-controller:devices" xmlns:clixon-controller="http://clicon.org/controller"/></get-config></rpc>"""
+
+    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"""<rpc xmlns="urn:ietf:params:xml:ns:netconf:base:1.0" username="{user}" xmlns:nc="urn:ietf:params:xml:ns:netconf:base:1.0" message-id="42"><get-config><source><actions xmlns="http://clicon.org/controller"/></source><nc:filter nc:type="xpath" nc:select="/clixon-controller:services/l2c:l2c[l2c:service-name=\'test\']" xmlns:clixon-controller="http://clicon.org/controller" xmlns:l2c="http://example.com/l2c" xmlns:custom="http://example.com/custom"/></get-config></rpc>"""
+
+    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