chore: minor cleanups from reverb (#618)

Signed-off-by: Henry Schreiner <[email protected]>

Signed-off-by: Henry Schreiner <[email protected]>

plumbum/cli/application.py

@@ -723,7 +723,7 @@ def cleanup(self, retcode):
     def helpall(self):
         """Prints help messages of all sub-commands and quits"""
         self.help()
-        print("")
+        print()
 
         if self._subcommands:
             for name, subcls in sorted(self._subcommands.items()):
@@ -744,7 +744,7 @@ def help(self):  # @ReservedAssignment
         """Prints this help message and quits"""
         if self._get_prog_version():
             self.version()
-            print("")
+            print()
         if self.DESCRIPTION:
             print(self.DESCRIPTION.strip() + "\n")
 
@@ -860,7 +860,7 @@ def wrapped_paragraphs(text, width):
             tailargs.append(f"{m.varargs}...")
         tailargs = " ".join(tailargs)
 
-        utc = self.COLOR_USAGE_TITLE if self.COLOR_USAGE_TITLE else self.COLOR_USAGE
+        utc = self.COLOR_USAGE_TITLE or self.COLOR_USAGE
         print(utc | T_("Usage:"))
 
         with self.COLOR_USAGE:
@@ -908,7 +908,7 @@ def switchs(by_groups, show_groups):
                     yield si, prefix, self.COLOR_GROUPS[grp]
 
                 if show_groups:
-                    print("")
+                    print()
 
         sw_width = (
             max(len(prefix) for si, prefix, color in switchs(by_groups, False)) + 4
@@ -954,14 +954,14 @@ def switchs(by_groups, show_groups):
             for name, subcls in sorted(self._subcommands.items()):
                 with gc:
                     subapp = subcls.get()
-                    doc = subapp.DESCRIPTION if subapp.DESCRIPTION else getdoc(subapp)
+                    doc = subapp.DESCRIPTION or getdoc(subapp)
                     if self.SUBCOMMAND_HELPMSG:
                         help_str = doc + "; " if doc else ""
                         help_str += self.SUBCOMMAND_HELPMSG.format(
                             parent=self.PROGNAME, sub=name
                         )
                     else:
-                        help_str = doc if doc else ""
+                        help_str = doc or ""
 
                     msg = indentation.join(
                         wrapper.wrap(

plumbum/cli/config.py

@@ -1,3 +1,4 @@
+import contextlib
 from abc import ABC, abstractmethod
 from configparser import ConfigParser, NoOptionError, NoSectionError
 
@@ -26,10 +27,8 @@ def __init__(self, filename):
         self.changed = False
 
     def __enter__(self):
-        try:
+        with contextlib.suppress(FileNotFoundError):
             self.read()
-        except FileNotFoundError:
-            pass
         return self
 
     def __exit__(self, exc_type, exc_val, exc_tb):

plumbum/cli/switches.py

@@ -1,3 +1,4 @@
+import contextlib
 import inspect
 from abc import ABC, abstractmethod
 
@@ -484,10 +485,8 @@ def __call__(self, value, check_csv=True):
                 if opt == value:
                     return opt  # always return original value
                 continue
-            try:
+            with contextlib.suppress(ValueError):
                 return opt(value)
-            except ValueError:
-                pass
         raise ValueError(f"Invalid value: {value} (Expected one of {self.values})")
 
     def choices(self, partial=""):

plumbum/cli/terminal.py

@@ -4,37 +4,39 @@
 """
 
 
+import contextlib
 import os
 import sys
+from typing import List, Optional
 
 from plumbum import local
 
 from .progress import Progress
 from .termsize import get_terminal_size
 
-__all__ = (
+__all__ = [
     "readline",
     "ask",
     "choose",
     "prompt",
     "get_terminal_size",
     "Progress",
     "get_terminal_size",
-)
+]
 
 
-def __dir__():
+def __dir__() -> List[str]:
     return __all__
 
 
-def readline(message=""):
+def readline(message: str = "") -> str:
     """Gets a line of input from the user (stdin)"""
     sys.stdout.write(message)
     sys.stdout.flush()
     return sys.stdin.readline()
 
 
-def ask(question, default=None):
+def ask(question: str, default: Optional[bool] = None) -> bool:
     """
     Presents the user with a yes/no question.
 
@@ -234,13 +236,9 @@ def pager(rows, pagercmd=None):  # pragma: no cover
         pg.stdin.close()
         pg.wait()
     finally:
-        try:
+        with contextlib.suppress(Exception):
             rows.close()
-        except Exception:
-            pass
         if pg and pg.poll() is None:
-            try:
+            with contextlib.suppress(Exception):
                 pg.terminate()
-            except Exception:
-                pass
             os.system("reset")

plumbum/cli/termsize.py

@@ -3,6 +3,7 @@
 ---------------------
 """
 
+import contextlib
 import os
 import platform
 import warnings
@@ -87,12 +88,10 @@ def _ioctl_GWINSZ(fd: int) -> Optional[Tuple[int, int]]:
 def _get_terminal_size_linux() -> Optional[Tuple[int, int]]:
     cr = _ioctl_GWINSZ(0) or _ioctl_GWINSZ(1) or _ioctl_GWINSZ(2)
     if not cr:
-        try:
+        with contextlib.suppress(Exception):
             fd = os.open(os.ctermid(), os.O_RDONLY)
             cr = _ioctl_GWINSZ(fd)
             os.close(fd)
-        except Exception:
-            pass
     if not cr:
         try:
             cr = (int(os.environ["LINES"]), int(os.environ["COLUMNS"]))

plumbum/colorlib/factories.py

@@ -3,8 +3,9 @@
 """
 
 
+import functools
+import operator
 import sys
-from functools import reduce
 from typing import Any
 
 from .names import color_names, default_styles
@@ -182,7 +183,7 @@ def get_colors_from_string(self, color=""):
                 prev = self
 
         if styleslist:
-            prev = reduce(lambda a, b: a & b, styleslist)
+            prev = functools.reduce(operator.and_, styleslist)
 
         return prev if isinstance(prev, self._style) else prev.reset
 

plumbum/colorlib/styles.py

@@ -8,6 +8,7 @@
 """
 
 
+import contextlib
 import os
 import platform
 import re
@@ -187,12 +188,10 @@ def from_simple(cls, color, fg=True):
         return self
 
     def _from_simple(self, color):
-        try:
+        with contextlib.suppress(AttributeError):
             color = color.lower()
             color = color.replace(" ", "")
             color = color.replace("_", "")
-        except AttributeError:
-            pass
 
         if color == "reset":
             return
@@ -219,12 +218,10 @@ def from_full(cls, color, fg=True):
         return self
 
     def _from_full(self, color):
-        try:
+        with contextlib.suppress(AttributeError):
             color = color.lower()
             color = color.replace(" ", "")
             color = color.replace("_", "")
-        except AttributeError:
-            pass
 
         if color == "reset":
             return
@@ -567,9 +564,9 @@ def __repr__(self):
         neg_attributes = ", ".join(
             f"-{a}" for a in self.attributes if not self.attributes[a]
         )
-        colors = ", ".join(repr(c) for c in [self.fg, self.bg] if c)
+        colors = ", ".join(repr(c) for c in (self.fg, self.bg) if c)
         string = (
-            "; ".join(s for s in [attributes, neg_attributes, colors] if s) or "empty"
+            "; ".join(s for s in (attributes, neg_attributes, colors) if s) or "empty"
         )
         if self.isreset:
             string = "reset"

plumbum/commands/base.py

@@ -1,7 +1,7 @@
+import contextlib
 import functools
 import shlex
 import subprocess
-from contextlib import contextmanager
 from subprocess import PIPE, Popen
 from tempfile import TemporaryFile
 from types import MethodType
@@ -172,7 +172,7 @@ def nohup(self, cwd=".", stdout="nohup.out", stderr=None, append=True):
         """Runs a command detached."""
         return self.machine.daemonic_popen(self, cwd, stdout, stderr, append)
 
-    @contextmanager
+    @contextlib.contextmanager
     def bgrun(self, args=(), **kwargs):
         """Runs the given command as a context manager, allowing you to create a
         `pipeline <http://en.wikipedia.org/wiki/Pipeline_(computing)>`_ (not in the UNIX sense)
@@ -215,11 +215,9 @@ def runner():
                 return run_proc(p, retcode, timeout)
             finally:
                 del p.run  # to break cyclic reference p -> cell -> p
-                for f in [p.stdin, p.stdout, p.stderr]:
-                    try:
+                for f in (p.stdin, p.stdout, p.stderr):
+                    with contextlib.suppress(Exception):
                         f.close()
-                    except Exception:
-                        pass
 
         p.run = runner
         yield p

plumbum/commands/daemons.py

@@ -1,3 +1,4 @@
+import contextlib
 import errno
 import os
 import signal
@@ -63,10 +64,8 @@ def posix_daemonize(command, cwd, stdout=None, stderr=None, append=True):
     _, rc = os.waitpid(firstpid, 0)
     output = os.read(rfd, MAX_SIZE)
     os.close(rfd)
-    try:
+    with contextlib.suppress(UnicodeError):
         output = output.decode("utf8")
-    except UnicodeError:
-        pass
     if rc == 0 and output.isdigit():
         secondpid = int(output)
     else:

plumbum/commands/processes.py

@@ -1,4 +1,5 @@
 import atexit
+import contextlib
 import heapq
 import math
 import time
@@ -220,26 +221,22 @@ def _timeout_thread_func():
                 timeout = max(0, ttk - time.time())
             else:
                 timeout = None
-            try:
+            with contextlib.suppress(QueueEmpty):
                 proc, time_to_kill = _timeout_queue.get(timeout=timeout)
                 if proc is SystemExit:
                     # terminate
                     return
                 waiting.push((time_to_kill, proc))
-            except QueueEmpty:
-                pass
             now = time.time()
             while waiting:
                 ttk, proc = waiting.peek()
                 if ttk > now:
                     break
                 waiting.pop()
-                try:
+                with contextlib.suppress(OSError):
                     if proc.poll() is None:
                         proc.kill()
                         proc._timed_out = True
-                except OSError:
-                    pass
     except Exception:
         if _shutting_down:
             # to prevent all sorts of exceptions during interpreter shutdown

plumbum/fs/atomic.py

@@ -3,9 +3,9 @@
 """
 
 import atexit
+import contextlib
 import os
 import threading
-from contextlib import contextmanager
 
 from plumbum.machines.local import local
 
@@ -24,7 +24,7 @@
         )
         raise
 
-    @contextmanager
+    @contextlib.contextmanager
     def locked_file(fileno, blocking=True):
         hndl = msvcrt.get_osfhandle(fileno)
         try:
@@ -46,7 +46,7 @@ def locked_file(fileno, blocking=True):
 else:
     if hasattr(fcntl, "lockf"):
 
-        @contextmanager
+        @contextlib.contextmanager
         def locked_file(fileno, blocking=True):
             fcntl.lockf(fileno, fcntl.LOCK_EX | (0 if blocking else fcntl.LOCK_NB))
             try:
@@ -56,7 +56,7 @@ def locked_file(fileno, blocking=True):
 
     else:
 
-        @contextmanager
+        @contextlib.contextmanager
         def locked_file(fileno, blocking=True):
             fcntl.flock(fileno, fcntl.LOCK_EX | (0 if blocking else fcntl.LOCK_NB))
             try:
@@ -114,7 +114,7 @@ def reopen(self):
             os.open(str(self.path), os.O_CREAT | os.O_RDWR, 384), "r+b", 0
         )
 
-    @contextmanager
+    @contextlib.contextmanager
     def locked(self, blocking=True):
         """
         A context manager that locks the file; this function is reentrant by the thread currently
@@ -274,10 +274,8 @@ def __exit__(self, t, v, tb):
         self.release()
 
     def __del__(self):
-        try:
+        with contextlib.suppress(Exception):
             self.release()
-        except Exception:  # pylint:disable=broad-except
-            pass
 
     def close(self):
         self.atomicfile.close()