summaryrefslogtreecommitdiff
path: root/python
diff options
context:
space:
mode:
authorJohn Estabrook <jestabro@vyos.io>2025-08-06 09:49:52 -0500
committerJohn Estabrook <jestabro@vyos.io>2025-08-06 09:49:52 -0500
commit19e5e463833aca9d4cf3494e422a1c2effd1412a (patch)
tree2571a991d16ba7ed6c25090f0ab7e98aa743201b /python
parent3f37c1dfaac577a46bb3c045e8ea3d1bc9420851 (diff)
downloadvyos-1x-19e5e463833aca9d4cf3494e422a1c2effd1412a.tar.gz
vyos-1x-19e5e463833aca9d4cf3494e422a1c2effd1412a.zip
T7693: formatting
Diffstat (limited to 'python')
-rw-r--r--python/vyos/configquery.py91
1 files changed, 61 insertions, 30 deletions
diff --git a/python/vyos/configquery.py b/python/vyos/configquery.py
index 9578fb627..b661b1a9e 100644
--- a/python/vyos/configquery.py
+++ b/python/vyos/configquery.py
@@ -13,10 +13,10 @@
# You should have received a copy of the GNU Lesser General Public License
# along with this library. If not, see <http://www.gnu.org/licenses/>.
-'''
+"""
A small library that allows querying existence or value(s) of config
settings from op mode, and execution of arbitrary op mode commands.
-'''
+"""
import os
import json
@@ -42,9 +42,11 @@ from vyos.configsource import ConfigSourceVyconfSession
config_file = os.path.join(directories['config'], 'config.boot')
+
class ConfigQueryError(Exception):
pass
+
class GenericConfigQuery:
def __init__(self):
pass
@@ -58,6 +60,7 @@ class GenericConfigQuery:
def values(self, path: list):
raise NotImplementedError
+
class GenericOpRun:
def __init__(self):
pass
@@ -65,6 +68,7 @@ class GenericOpRun:
def run(self, path: list, **kwargs):
raise NotImplementedError
+
class CliShellApiConfigQuery(GenericConfigQuery):
def __init__(self):
super().__init__()
@@ -90,6 +94,7 @@ class CliShellApiConfigQuery(GenericConfigQuery):
raise ConfigQueryError('No values for given path')
return out
+
class ConfigTreeQuery(GenericConfigQuery):
def __init__(self):
super().__init__()
@@ -104,11 +109,12 @@ class ConfigTreeQuery(GenericConfigQuery):
try:
with open(config_file) as f:
config_string = f.read()
- except OSError as err:
+ except OSError:
config_string = ''
- config_source = ConfigSourceString(running_config_text=config_string,
- session_config_text=config_string)
+ config_source = ConfigSourceString(
+ running_config_text=config_string, session_config_text=config_string
+ )
self.config = Config(config_source=config_source)
def exists(self, path: list):
@@ -123,16 +129,28 @@ class ConfigTreeQuery(GenericConfigQuery):
def list_nodes(self, path: list):
return self.config.list_nodes(path)
- def get_config_dict(self, path=[], effective=False, key_mangling=None,
- get_first_key=False, no_multi_convert=False,
- no_tag_node_value_mangle=False, with_defaults=False,
- with_recursive_defaults=False):
- return self.config.get_config_dict(path, effective=effective,
- key_mangling=key_mangling, get_first_key=get_first_key,
- no_multi_convert=no_multi_convert,
- no_tag_node_value_mangle=no_tag_node_value_mangle,
- with_defaults=with_defaults,
- with_recursive_defaults=with_recursive_defaults)
+ def get_config_dict(
+ self,
+ path=[],
+ effective=False,
+ key_mangling=None,
+ get_first_key=False,
+ no_multi_convert=False,
+ no_tag_node_value_mangle=False,
+ with_defaults=False,
+ with_recursive_defaults=False,
+ ):
+ return self.config.get_config_dict(
+ path,
+ effective=effective,
+ key_mangling=key_mangling,
+ get_first_key=get_first_key,
+ no_multi_convert=no_multi_convert,
+ no_tag_node_value_mangle=no_tag_node_value_mangle,
+ with_defaults=with_defaults,
+ with_recursive_defaults=with_recursive_defaults,
+ )
+
class VbashOpRun(GenericOpRun):
def __init__(self):
@@ -140,35 +158,45 @@ class VbashOpRun(GenericOpRun):
def run(self, path: list, **kwargs):
cmd = ' '.join(path)
- (out, err) = popen(f'/opt/vyatta/bin/vyatta-op-cmd-wrapper {cmd}', stderr=STDOUT, **kwargs)
+ (out, err) = popen(
+ f'/opt/vyatta/bin/vyatta-op-cmd-wrapper {cmd}', stderr=STDOUT, **kwargs
+ )
if err:
raise ConfigQueryError(out)
return out
-def query_context(config_query_class=CliShellApiConfigQuery,
- op_run_class=VbashOpRun):
+
+def query_context(config_query_class=CliShellApiConfigQuery, op_run_class=VbashOpRun):
query = config_query_class()
run = op_run_class()
return query, run
+
def verify_mangling(key_mangling):
- if not (isinstance(key_mangling, tuple) and
- len(key_mangling) == 2 and
- isinstance(key_mangling[0], str) and
- isinstance(key_mangling[1], str)):
- raise ValueError("key_mangling must be a tuple of two strings")
+ if not (
+ isinstance(key_mangling, tuple)
+ and len(key_mangling) == 2
+ and isinstance(key_mangling[0], str)
+ and isinstance(key_mangling[1], str)
+ ):
+ raise ValueError('key_mangling must be a tuple of two strings')
+
def op_mode_run(cmd):
- """ low-level to avoid overhead """
+ """low-level to avoid overhead"""
p = subprocess.Popen(cmd, stdout=subprocess.PIPE)
out = p.stdout.read()
p.wait()
return p.returncode, out.decode()
-def op_mode_config_dict(path=None, key_mangling=None,
- no_tag_node_value_mangle=False,
- no_multi_convert=False, get_first_key=False):
+def op_mode_config_dict(
+ path=None,
+ key_mangling=None,
+ no_tag_node_value_mangle=False,
+ no_multi_convert=False,
+ get_first_key=False,
+):
if path is None:
path = []
command = ['/bin/cli-shell-api', '--show-active-only', 'showConfig']
@@ -193,8 +221,11 @@ def op_mode_config_dict(path=None, key_mangling=None,
if key_mangling is not None:
verify_mangling(key_mangling)
- config_dict = mangle_dict_keys(config_dict,
- key_mangling[0], key_mangling[1],
- no_tag_node_value_mangle=no_tag_node_value_mangle)
+ config_dict = mangle_dict_keys(
+ config_dict,
+ key_mangling[0],
+ key_mangling[1],
+ no_tag_node_value_mangle=no_tag_node_value_mangle,
+ )
return get_sub_dict(config_dict, path, get_first_key=get_first_key)