More f-string conversion

- Move lots more str.format() calls to f-strings.
- Minor typing and comment improvements

docs/source/dyn/cli.rst

@@ -372,7 +372,8 @@ ksconf filter
       Include or exclude entire stanzas using these filter options. All filter
       options can be provided multiple times. If you have a long list of
       filters, they can be saved in a file and referenced using the special
-      'file://' prefix. One entry per line.
+      'file://' prefix. One entry per line. Entries can be either a literal
+      strings, wildcards, or regexes, depending on MATCH.
     
       --stanza PATTERN      Match any stanza who's name matches the given pattern.
                             PATTERN supports bulk patterns via the 'file://'

ksconf/commands/filter.py

@@ -99,7 +99,8 @@ def register_args(self, parser: ArgumentParser):
 
             All filter options can be provided multiple times.
             If you have a long list of filters, they can be saved in a file and referenced using
-            the special ``file://`` prefix.  One entry per line."""))
+            the special ``file://`` prefix.  One entry per line.  Entries can be either a
+            literal strings, wildcards, or regexes, depending on MATCH."""))
 
         pg_sel.add_argument("--stanza", metavar="PATTERN", action="append", default=[],
                             help=dedent("""
@@ -184,7 +185,7 @@ def prep_filters(self, args):
             # Bypass filter
             self.filter_attrs = lambda x: x
 
-    def _test_stanza(self, stanza, attributes):
+    def _test_stanza(self, stanza: str, attributes: dict) -> bool:
         if self.stanza_filters.match_stanza(stanza):
             # Exclude based on value of 'disabled' attribute
             if not self.disabled_filter(attributes):
@@ -199,40 +200,43 @@ def _test_stanza(self, stanza, attributes):
                     return True
         return False
 
-    def filter_attrs(self, content):
+    def filter_attrs(self, content: dict) -> dict:
         d = {}
         for (attr, value) in content.items():
             if self.attrs_keep_filter.match(attr) and self.attrs_reject_filter.match(attr):
                 d[attr] = content[attr]
         return d
 
-    def output(self, args, matches, filename):
+    def output(self, args, matches: dict, filename):
+        """
+        Process output for a single input file.
+        """
         if args.files_with_matches:
             if matches:
                 if args.count:
-                    args.output.write("{} has {} matching stanza(s)\n".format(filename, len(matches)))
+                    args.output.write(f"{filename} has {len(matches)} matching stanza(s)\n")
                 elif args.brief:
                     for stanza_name in matches:
-                        args.output.write("{}: {}\n".format(filename, stanza_name))
+                        args.output.write(f"{filename}: {stanza_name}\n")
                 else:
                     # Just show a single file
-                    args.output.write("{}\n".format(filename))
+                    args.output.write(f"{filename}\n")
             elif args.verbose:
-                self.stderr.write("No matching stanzas in {}\n".format(filename))
+                self.stderr.write(f"No matching stanzas in {filename}\n")
         elif args.count:
-            args.output.write("{}\n".format(len(matches)))
+            args.output.write(f"{len(matches)}\n")
         elif args.brief:
             for stanza_name in matches:
-                args.output.write("{}\n".format(stanza_name))
+                args.output.write(f"{stanza_name}\n")
         else:
             if len(args.conf) > 1:
-                args.output.write("#  {}\n".format(filename))
+                args.output.write(f"#  {filename}\n")
             if matches:
                 write_conf_stream(args.output, matches)
             elif args.verbose:
-                self.stderr.write("No matching stanzas in {}\n".format(filename))
+                self.stderr.write(f"No matching stanzas in {filename}\n")
             if args.verbose:
-                sys.stderr.write("Matched {} stanzas from {}\n".format(len(matches), filename))
+                sys.stderr.write(f"Matched {len(matches)} stanzas from {filename}\n")
 
     def run(self, args):
         ''' Filter configuration files. '''
@@ -247,7 +251,6 @@ def run(self, args):
         for conf in args.conf:
             conf.set_parser_option(keep_comments=args.comments)
             cfg = conf.data
-            # Should this be an ordered dict?
             cfg_out = dict()
             for stanza_name, attributes in cfg.items():
                 keep = self._test_stanza(stanza_name, attributes) ^ args.invert_match

ksconf/commands/package.py

@@ -169,8 +169,8 @@ def pre_run(self, args):
             if pattern.startswith("file://"):
                 blocklist_file = pattern[7:]
                 expanded = list(self.load_blocklist(blocklist_file))
-                self.stderr.write("Extended blocklist from {} with {:d} entries\n"
-                                  .format(blocklist_file, len(expanded)))
+                self.stderr.write(f"Extended blocklist from {blocklist_file} "
+                                  f"with {len(expanded)} entries\n")
                 blocklist.extend(expanded)
             else:
                 blocklist.append(pattern)

ksconf/commands/promote.py

@@ -410,9 +410,9 @@ def prompt_yes_no(prompt):
                 # Q:  Should we simply remove everything from the source file that already lines
                 #     up with the target?  Just ask
                 if isinstance(op.location, DiffStzKey):
-                    msg = "[{0.stanza}]  {0.key}".format(op.location)
+                    msg = f"[{op.location.stanza}]  {op.location.key}"
                 elif isinstance(op.location, DiffStanza):
-                    msg = "[{0.stanza}]".format(op.location)
+                    msg = f"[{op.location.stanza}]"
                 if prompt_yes_no(f"Remove matching entry {msg}  "):
                     if isinstance(op.location, DiffStanza):
                         del out_src[op.location.stanza]

ksconf/commands/restpublish.py

@@ -174,12 +174,11 @@ def handle_conf_file(self, args: Namespace, conf_proxy: ConfFileProxy):
             else:
                 action, info = self.publish_conf(stanza_name, stanza_data, config_file)
 
-            print("{:50} {:8}   (delta size: {})".format("[{}]".format(stanza_name),
-                                                         action, len(info.get("delta", []))))
+            print(f"{stanza_name:50} {action:8}   (delta size: { len(info.get('delta', [])) })")
 
             update_time = info.get("updated", 0)
-            # headers = (conf_proxy.name, "{}/{}".format(args.url, config_file.path))
-            # rest_header = DiffHeader("{}/{}".format(args.url, info.get("path", config_file.path), update_time))
+            # headers = (conf_proxy.name, f"{args.url}/{config_file.path}")
+            # rest_header = DiffHeader(f"{args.url}/{info.get('path', config_file.path)}", update_time)
             rest_header = DiffHeader(info.get("path", config_file.path), update_time)
             if action != "nochange" and "delta" in info:
                 show_diff(self.stdout, info["delta"], headers=(conf_proxy.name, rest_header))

ksconf/commands/sort.py

@@ -87,7 +87,7 @@ def run(self, args):
                 try:
                     if not args.force and _has_nosort_marker(conf):
                         if not args.quiet:
-                            self.stderr.write("Skipping no-sort file {}\n".format(conf))
+                            self.stderr.write(f"Skipping no-sort file {conf}\n")
                         continue
                     c = self.parse_conf(conf, mode='r+', raw_exec=True)
                     # c = parse_conf(conf, profile=PARSECONF_STRICT)
@@ -97,16 +97,16 @@ def run(self, args):
                     #                            sort=True)
                 except ConfParserException as e:
                     smart_rc = None
-                    self.stderr.write("Error trying to process file {0}.  "
-                                      "Error:  {1}\n".format(conf, e))
+                    self.stderr.write(f"Error trying to process file {conf}.  "
+                                      f"Error:  {e}\n")
                     failure = True
                     continue
                 if smart_rc == SMART_NOCHANGE:
                     if not args.quiet:
                         self.stderr.write("Nothing to update.  "
-                                          "File {0} is already sorted\n".format(conf))
+                                          f"File {conf} is already sorted\n")
                 else:
-                    self.stderr.write("Replaced file {0} with sorted content.\n".format(conf))
+                    self.stderr.write(f"Replaced file {conf} with sorted content.\n")
                     changes += 1
             if failure:
                 return EXIT_CODE_BAD_CONF_FILE
@@ -115,8 +115,7 @@ def run(self, args):
         else:
             for conf in args.conf:
                 if len(args.conf) > 1:
-                    args.target.write("---------------- [ {0} ] ----------------\n\n"
-                                      .format(conf))
+                    args.target.write(f"---------------- [ {conf} ] ----------------\n\n")
                 data = self.parse_conf(conf).data
                 write_conf(args.target, data, stanza_delim=stanza_delims, sort=True)
             return EXIT_CODE_SUCCESS

ksconf/commands/unarchive.py

@@ -282,7 +282,7 @@ def fixup_pattern_bw(patterns, prefix=None):
             for pattern in patterns:
                 if pattern.startswith("./"):
                     if prefix:
-                        pattern = "{0}/{1}".format(prefix, pattern[2:])
+                        pattern = f"{prefix}/{pattern[2:]}"
                     else:
                         pattern = pattern[2:]
                     modified.append(pattern)
@@ -300,16 +300,16 @@ def fixup_pattern_bw(patterns, prefix=None):
             for pattern in local_files:
                 excludes.append("./" + pattern)
         excludes = fixup_pattern_bw(excludes, app_basename)
-        self.stderr.write("Extraction exclude patterns:  {!r}\n".format(excludes))
+        self.stderr.write(f"Extraction exclude patterns:  {excludes!r}\n")
         exclude_filter = create_filtered_list("splunk", default=False)
         exclude_filter.feedall(excludes)
 
         # Calculate path rewrite operations
         path_rewrites = []
         files_iter = extract_archive(args.tarball)
         if args.default_dir != DEFAULT_DIR:
-            rep = r"\1/{}/".format(args.default_dir.strip("/"))
-            path_rewrites.append((re.compile(r"^(/?[^/]+)/{}/".format(DEFAULT_DIR)), rep))
+            rep = rf"\1/{ args.default_dir.strip('/') }/"
+            path_rewrites.append((re.compile(rf"^(/?[^/]+)/{DEFAULT_DIR}/"), rep))
             del rep
         if new_app_name:
             regex = re.compile(r'^([^/]+)(?=/)')
@@ -323,7 +323,7 @@ def fixup_pattern_bw(patterns, prefix=None):
         self.stdout.write("Extracting app now...\n")
         for gaf in files_iter:
             if exclude_filter.match(gaf.path):
-                self.stdout.write("Skipping [blocklist] {}\n".format(gaf.path))
+                self.stdout.write(f"Skipping [blocklist] {gaf.path}\n")
                 continue
             if not is_git or args.git_mode in ("nochange", "stage"):
                 self.stdout.write(f"{gaf.path:60s} {gaf.mode:o} {gaf.size:-6d}\n")
@@ -338,9 +338,12 @@ def fixup_pattern_bw(patterns, prefix=None):
         files_new, files_upd, files_del = cmp_sets(installed_files, existing_files)
 
         if DEBUG:
-            print("New: \n\t{}".format("\n\t".join(sorted(files_new))))
-            print("Existing: \n\t{}".format("\n\t".join(sorted(files_upd))))
-            print("Removed:  \n\t{}".format("\n\t".join(sorted(files_del))))
+            def dbg_fmt(l):
+                return "\n\t".join(sorted(l))
+
+            print(f"New:      \n\t{dbg_fmt(files_new)}")
+            print(f"Existing: \n\t{dbg_fmt(files_upd)}")
+            print(f"Removed:  \n\t{dbg_fmt(files_del)}")
 
         self.stdout.write(f"Extracted {len(installed_files)} files:  "
                           f"{len(files_new)} new, {len(files_upd)} existing, "
@@ -395,10 +398,10 @@ def fixup_pattern_bw(patterns, prefix=None):
         if is_git:
             if args.git_mode in ("stage", "commit"):
                 git_cmd(["add", "--all", dest_app.name], cwd=dest_app.parent)
-                # self.stdout.write("git add {}\n".format(os.path.basename(dest_app)))
+                # self.stdout.write(f"git add {os.path.basename(dest_app)}\n")
             '''
             else:
-                self.stdout.write("git add {}\n".format(dest_app))
+                self.stdout.write(f"git add {dest_app}\n")
             '''
 
             # Is there anything to stage/commit?

ksconf/commands/xmlformat.py

@@ -46,7 +46,7 @@ def _handle_imports(cls):
         if globals()["etree"]:
             return
         from lxml import etree
-        cls.version_extra = "lxml {}".format(etree.__version__)
+        cls.version_extra = f"lxml {etree.__version__}"
         g["etree"] = etree
 
     def register_args(self, parser):
@@ -118,26 +118,26 @@ def run(self, args):
         for fn in files:
             c["checked"] += 1
             if not os.path.isfile(fn):
-                self.stderr.write("Skipping missing file:  {0}\n".format(fn))
+                self.stderr.write(f"Skipping missing file:  {fn}\n")
                 c["missing"] += 1
                 continue
             try:
                 if formatter.format_xml(fn, fn, args.indent):
-                    self.stderr.write("Replaced file {0} with formatted content\n".format(fn))
+                    self.stderr.write(f"Replaced file {fn} with formatted content\n")
                     c["changed"] += 1
                 else:
                     if not args.quiet:
-                        self.stderr.write("Already formatted {0}\n".format(fn))
+                        self.stderr.write(f"Already formatted {fn}\n")
                     c["no-action"] += 1
                 self.stderr.flush()
             except etree.ParseError as e:
-                self.stderr.write("Error parsing file {0}:  {1}\n".format(fn, e))
+                self.stderr.write(f"Error parsing file {fn}:  {e}\n")
                 self.stderr.flush()
                 c["error"] += 1
                 exit_code = EXIT_CODE_BAD_CONF_FILE
             except Exception as e:  # pragma: no cover
-                self.stderr.write("Unhandled top-level exception while parsing {0}.  "
-                                  "Aborting.\n{1}\n".format(fn, e))
+                self.stderr.write(f"Unhandled top-level exception while parsing {fn}.  "
+                                  f"Aborting.\n{e}\n")
                 debug_traceback()
                 c["error"] += 1
                 exit_code = EXIT_CODE_INTERNAL_ERROR
@@ -147,8 +147,9 @@ def run(self, args):
             exit_code = EXIT_CODE_FORMAT_APPLIED
 
         if True:  # show stats or verbose
-            self.stdout.write("Completed formatting {0[checked]} files.  rc={1} Breakdown:\n"
-                              "   {0[changed]} files were formatted successfully.\n"
-                              "   {0[no-action]} files were already formatted.\n"
-                              "   {0[error]} files failed.\n".format(c, exit_code))
+            self.stdout.write(f"Completed formatting {c['checked']} files.  "
+                              f"rc={exit_code} Breakdown:\n"
+                              f"   {c['changed']} files were formatted successfully.\n"
+                              f"   {c['no-action']} files were already formatted.\n"
+                              f"   {c['error']} files failed.\n")
         return exit_code

ksconf/conf/delta.py

@@ -257,12 +257,11 @@ def summarize_cfg_diffs(delta: List[DiffOp], stream: TextIO):
             key_stats[op.tag][op.location.stanza][op.location.key].add(op.location.key)
 
     for tag in sorted(c.keys()):
-        stream.write("Have {0} '{1}' operations:\n".format(c[tag], tag))
+        stream.write(f"Have {c[tag]} '{tag}' operations:\n")
         for entry in sorted(stanza_stats[tag]):
-            stream.write("\t[{0}]\n".format(_format_stanza(entry)))
+            stream.write(f"\t[{_format_stanza(entry)}]\n")
         for entry in sorted(key_stats[tag]):
-            stream.write("\t[{0}]  {1} keys\n".format(_format_stanza(entry),
-                                                      len(key_stats[tag][entry])))
+            stream.write(f"\t[{_format_stanza(entry)}]  {len(key_stats[tag][entry])} keys\n")
         stream.write("\n")
 
 

ksconf/conf/merge.py

@@ -123,7 +123,7 @@ def merge_update_any_file(dest: str,
     # XXX: Set this up in a shared function somewhere, we have to figure this out multiple places
     remove = []
     if dest.endswith(".conf") or dest.endswith(".meta"):
-        # self.output.write("Merge   {} -> {}\n".format(local, default))
+        # self.output.write(f"Merge   {local} -> {default}\n")
         merge_update_conf_file(dest, sources, remove_source=remove_source)
         # merge_conf(default, local)
     else:

ksconf/conf/meta.py

@@ -18,6 +18,7 @@
 from __future__ import absolute_import, unicode_literals
 
 import re
+from typing import TextIO
 from urllib.parse import quote, unquote
 
 from ksconf.conf.parser import GLOBAL_STANZA, parse_conf
@@ -108,7 +109,7 @@ def parse_meta(cls, stanza):
         if "access" in stanza:
             access = stanza["access"]
             for match in re.finditer(cls.regex_access, access):
-                stanza_name = "access.{}".format(match.group("action"))
+                stanza_name = f"access.{ match.group('action') }"
                 values = [role.strip() for role in match.group("roles").split(",")]
                 stanza[stanza_name] = values
         return stanza
@@ -127,7 +128,7 @@ def get(self, *names):
             layers.append(node.data)
         d = self.expand_layers(layers)
         # Parser access:   split out 'read' vs 'write', return as list
-        # d["acesss"]
+        # d["access"]
         return self.parse_meta(d)
 
     def feed_file(self, stream):
@@ -153,7 +154,7 @@ def walk(self):
         for path in self._meta.walk():
             yield (path, self.get(*path))
 
-    def write_stream(self, stream, sort=True):
+    def write_stream(self, stream: TextIO, sort=True):
         if sort:
             # Prefix level # to list for sorting purposes
             data = [(len(parts), parts, payload) for parts, payload in self.iter_raw()]
@@ -165,12 +166,12 @@ def write_stream(self, stream, sort=True):
 
         for parts, payload in raw:
             stanza = "/".join(quote(p, "") for p in parts)
-            stream.write("[{}]\n".format(stanza))
+            stream.write(f"[{stanza}]\n")
             for attr in sorted(payload):
                 value = payload[attr]
                 if attr.startswith("#"):
-                    stream.write("{0}\n".format(value))
+                    stream.write(f"{value}\n")
                 else:
-                    stream.write("{} = {}\n".format(attr, value))
+                    stream.write(f"{attr} = {value}\n")
             # Trailing EOL, oh well...  fix that later
             stream.write("\n")

ksconf/filter.py

@@ -196,7 +196,7 @@ def _pre_match(self):
 }
 
 
-def create_filtered_list(match_mode, flags=0, default=True):
+def create_filtered_list(match_mode: str, flags=0, default=True) -> FilteredList:
     try:
         class_ = class_mapping[match_mode]
     except KeyError:

ksconf/util/file.py

@@ -96,17 +96,17 @@ def splglob_to_regex(pattern, re_flags=None):
     regex = re.sub(_glob_to_regex_find, lambda m: glob_to_regex[m.group()], pattern)
     # If NO anchors have been explicitly given, then assume full-match mode:
     if not re.search(r'(?<![\[\\])[$^]', regex):
-        regex = "^{}$".format(regex)
+        regex = f"^{regex}$"
     return re.compile(regex, flags=re_flags)
 
 
 def splglob_simple(pattern):
     """ Return a splglob that either matches a full path or match a simple file """
     if "/" not in pattern:
         # Assume we've been given a simple file name:   app.conf, *.tgz
-        pattern = "^.../{}$".format(pattern)
+        pattern = f"^.../{pattern}$"
     else:
-        pattern = "^{}$".format(pattern)
+        pattern = f"^{pattern}$"
     return pattern