feat: Allow for filtering of output (#174)

* feat: Allow for filtering of output

  - Command output may be filtered by `|` followed by whitespace and a regular expression (or a simple string).
  - The regular expression is applied to each line of output, and only lines that match are displayed.
  - The expression may be negated by using `|!` instead of `|`.
  - Substitution is not supported.

Examples:

- `network list_used_addresses 172.16.20.0 | bl23-.*mu2[17]` will output only lines matching the regular expression.
- `network list_used_addresses 172.16.20.0 |! mu2[17]` will output only lines *not* matching the regular expression.

This applies all commands that output to the console, but it does *not* apply to errors, warnings, or informational messages. This allows filters to be used without concern for missing important information.

This is a cleaner Singleton-based implementation of #171.

Author: terjekv

mreg_cli/bacnet.py

@@ -2,34 +2,28 @@
 from .history import history
 from .host import host
 from .log import cli_error, cli_info
-from .util import delete, get, get_list, host_info_by_name, post, print_table
+from .util import add_formatted_table_for_output, delete, get, get_list, host_info_by_name, post
 
 
-def bacnetid_add(args):
+def bacnetid_add(args) -> None:
     info = host_info_by_name(args.name)
     if "bacnetid" in info and info["bacnetid"] is not None:
-        cli_error(
-            "{} already has BACnet ID {}.".format(info["name"], info["bacnetid"]["id"])
-        )
+        cli_error("{} already has BACnet ID {}.".format(info["name"], info["bacnetid"]["id"]))
     postdata = {"hostname": info["name"]}
     path = "/api/v1/bacnet/ids/"
     bacnetid = args.id
     if bacnetid:
         response = get(path + bacnetid, ok404=True)
         if response:
             j = response.json()
-            cli_error(
-                "BACnet ID {} is already in use by {}".format(j["id"], j["hostname"])
-            )
+            cli_error("BACnet ID {} is already in use by {}".format(j["id"], j["hostname"]))
         postdata["id"] = bacnetid
     history.record_post(path, "", postdata)
     post(path, **postdata)
     info = host_info_by_name(args.name)
     if "bacnetid" in info and info["bacnetid"] is not None:
         b = info["bacnetid"]
-        cli_info(
-            "Assigned BACnet ID {} to {}".format(b["id"], info["name"]), print_msg=True
-        )
+        cli_info("Assigned BACnet ID {} to {}".format(b["id"], info["name"]), print_msg=True)
 
 
 host.add_command(
@@ -44,7 +38,7 @@ def bacnetid_add(args):
 )
 
 
-def bacnetid_remove(args):
+def bacnetid_remove(args) -> None:
     info = host_info_by_name(args.name)
     if "bacnetid" not in info or info["bacnetid"] is None:
         cli_error("{} does not have a BACnet ID assigned.".format(info["name"]))
@@ -68,7 +62,7 @@ def bacnetid_remove(args):
 )
 
 
-def bacnetid_list(args):
+def bacnetid_list(args) -> None:
     minval = 0
     if args.min is not None:
         minval = args.min
@@ -80,7 +74,7 @@ def bacnetid_list(args):
         if maxval > 4194302:
             cli_error("The maximum ID value is 4194302.")
     r = get_list("/api/v1/bacnet/ids/", {"id__range": "{},{}".format(minval, maxval)})
-    print_table(("ID", "Hostname"), ("id", "hostname"), r)
+    add_formatted_table_for_output(("ID", "Hostname"), ("id", "hostname"), r)
 
 
 host.add_command(

mreg_cli/cli.py

@@ -1,5 +1,7 @@
 import argparse
+import html
 import os
+import shlex
 
 from prompt_toolkit import HTML, print_formatted_text
 from prompt_toolkit.completion import Completer, Completion
@@ -90,9 +92,7 @@ def add_command(
             flags = []
         if not self.sub:
             self.sub = _create_command_group(self.parser)
-        parser = self.sub.add_parser(
-            prog, description=description, epilog=epilog, help=short_desc
-        )
+        parser = self.sub.add_parser(prog, description=description, epilog=epilog, help=short_desc)
         for f in flags:
             # Need to create a dict with the parameters so only used
             # parameters are sent, or else exceptions are raised. Ex: if
@@ -122,7 +122,11 @@ def add_command(
         self.children[prog] = new_cmd
         return new_cmd
 
-    def parse(self, args):
+    def parse(self, command: str) -> None:
+        """Parse and execute a command."""
+
+        args = shlex.split(command, comments=True)
+
         try:
             args = self.parser.parse_args(args)
             # If the command has a callback function, call it.
@@ -251,8 +255,6 @@ def source(files, ignore_errors, verbose):
     newlines.
     The files may contain comments. The comment symbol is #.
     """
-    import html
-    import shlex
 
     rec = recorder.Recorder()
 
@@ -272,16 +274,10 @@ def source(files, ignore_errors, verbose):
                     if rec.is_recording() and not line.lstrip().startswith("source"):
                         rec.record_command(line)
 
-                    # With comments=True shlex will remove comments from the line
-                    # when splitting. Comment symbol is #
-                    s = shlex.split(line, comments=True)
-
                     # In verbose mode all commands are printed before execution.
-                    if verbose and s:
-                        print_formatted_text(
-                            HTML(f"<i>> {html.escape(line.strip())}</i>")
-                        )
-                    cli.parse(s)
+                    if verbose:
+                        print_formatted_text(HTML(f"<i>> {html.escape(line.strip())}</i>"))
+                    cli.parse(line)
                     if cli.last_errno != 0:
                         print_formatted_text(
                             HTML(
@@ -324,8 +320,7 @@ def _source(args):
         Flag(
             "-ignore-errors",
             description=(
-                "Continue command execution on error. Default is to "
-                "stop execution on error."
+                "Continue command execution on error. Default is to stop execution on error."
             ),
             short_desc="Stop on error.",
             action="store_true",

mreg_cli/dhcp.py

@@ -90,7 +90,7 @@ def assoc_mac_to_ip(mac, ip, force=False):
 #########################################
 
 
-def assoc(args):
+def assoc(args) -> None:
     # .name .mac .force
     """Associate MAC address with host. If host got multiple A/AAAA records an
     IP must be given instead of name.
@@ -107,8 +107,10 @@ def assoc(args):
 
 dhcp.add_command(
     prog="assoc",
-    description="Associate MAC address with host. If host got multiple A/AAAA "
-    "records an IP must be given instead of name.",
+    description=(
+        "Associate MAC address with host. If host got multiple A/AAAA "
+        "records an IP must be given instead of name."
+    ),
     short_desc="Add MAC address to host.",
     callback=assoc,
     flags=[
@@ -124,7 +126,7 @@ def assoc(args):
 ############################################
 
 
-def disassoc(args):
+def disassoc(args) -> None:
     """Disassociate MAC address with host/ip. If host got multiple A/AAAA
     records an IP must be given instead of name.
     """
@@ -148,8 +150,10 @@ def disassoc(args):
 
 dhcp.add_command(
     prog="disassoc",
-    description="Disassociate MAC address with host/ip. If host got multiple "
-    "A/AAAA records an IP must be given instead of name.",
+    description=(
+        "Disassociate MAC address with host/ip. If host got multiple "
+        "A/AAAA records an IP must be given instead of name."
+    ),
     short_desc="Disassociate MAC address.",
     callback=disassoc,
     flags=[