diff options
| author | John Estabrook <jestabro@vyos.io> | 2025-08-06 09:49:52 -0500 |
|---|---|---|
| committer | John Estabrook <jestabro@vyos.io> | 2025-08-06 09:49:52 -0500 |
| commit | 19e5e463833aca9d4cf3494e422a1c2effd1412a (patch) | |
| tree | 2571a991d16ba7ed6c25090f0ab7e98aa743201b /python | |
| parent | 3f37c1dfaac577a46bb3c045e8ea3d1bc9420851 (diff) | |
| download | vyos-1x-19e5e463833aca9d4cf3494e422a1c2effd1412a.tar.gz vyos-1x-19e5e463833aca9d4cf3494e422a1c2effd1412a.zip | |
T7693: formatting
Diffstat (limited to 'python')
| -rw-r--r-- | python/vyos/configquery.py | 91 |
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) |
