Refactor and simplify code printing to stderr. (#762)

signac/__main__.py

@@ -34,7 +34,7 @@
 from .common.configobj import Section, flatten_errors
 from .contrib.filterparse import _add_prefix, parse_filter_arg
 from .contrib.import_export import _SchemaPathEvaluationError, export_jobs
-from .contrib.utility import _add_verbosity_argument, _query_yes_no
+from .contrib.utility import _add_verbosity_argument, _print_err, _query_yes_no
 from .diff import diff_jobs
 from .errors import (
     DestinationExistsError,
@@ -97,10 +97,6 @@
 warnings.simplefilter("default")
 
 
-def _print_err(msg=None, *args):
-    print(msg, *args, file=sys.stderr)
-
-
 def _fmt_bytes(nbytes, suffix="B"):
     """Format number of bytes.
 
@@ -1664,7 +1660,6 @@ def main():
     try:
         args.func(args)
     except KeyboardInterrupt:
-        _print_err()
         _print_err("Interrupted.")
         if args.debug:
             raise

signac/contrib/filterparse.py

@@ -4,40 +4,9 @@
 """Parse the filter arguments."""
 
 import json
-import sys
 from collections.abc import Mapping
 
-
-def _print_err(msg=None):
-    """Print the provided message to stderr.
-
-    Parameters
-    ----------
-    msg : str
-        Error message to be printed (Default value = None).
-
-    """
-    print(msg, file=sys.stderr)
-
-
-def _with_message(query, file):
-    """Print the interpreted filter arguments to the provided file.
-
-    Parameters
-    ----------
-    query : dict
-        Filter arguments.
-    file :
-        The file where the filter interpretation is printed.
-
-    Returns
-    -------
-    query : dict
-        Filter arguments.
-
-    """
-    print(f"Interpreted filter arguments as '{json.dumps(query)}'.", file=file)
-    return query
+from .utility import _print_err
 
 
 def _is_json_like(q):
@@ -125,7 +94,7 @@ def _cast(x):
     """
     try:
         if x in CAST_MAPPING_WARNING:
-            print(f"Did you mean {CAST_MAPPING_WARNING[x]}?", file=sys.stderr)
+            _print_err(f"Did you mean {CAST_MAPPING_WARNING[x]}?")
         return CAST_MAPPING[x]
     except KeyError:
         try:
@@ -196,15 +165,13 @@ def parse_simple(tokens):
         yield _parse_single(key, value)
 
 
-def parse_filter_arg(args, file=sys.stderr):
+def parse_filter_arg(args):
     """Parse a series of filter arguments into a dictionary.
 
     Parameters
     ----------
     args : sequence of str
         Filter arguments to parse.
-    file :
-        The file to write message (Default value = sys.stderr).
 
     Returns
     -------
@@ -219,10 +186,11 @@ def parse_filter_arg(args, file=sys.stderr):
             return _parse_json(args[0])
         else:
             key, value = _parse_single(args[0])
-            return _with_message({key: value}, file)
+            query = {key: value}
     else:
-        q = dict(parse_simple(args))
-        return _with_message(q, file)
+        query = dict(parse_simple(args))
+    _print_err(f"Interpreted filter arguments as '{json.dumps(query)}'.")
+    return query
 
 
 def _add_prefix(prefix, filter):

signac/contrib/migration/__init__.py

@@ -4,11 +4,11 @@
 """Handle migrations of signac schema versions."""
 
 import os
-import sys
 
 from filelock import FileLock
 
 from ...version import SCHEMA_VERSION, __version__
+from ..utility import _print_err
 from .v0_to_v1 import _load_config_v1, _migrate_v0_to_v1
 from .v1_to_v2 import _load_config_v2, _migrate_v1_to_v2
 
@@ -110,10 +110,9 @@ def apply_migrations(root_directory):
         with lock:
             for (origin, destination), migrate in _collect_migrations(root_directory):
                 try:
-                    print(
+                    _print_err(
                         f"Applying migration for version {origin} to {destination}... ",
                         end="",
-                        file=sys.stderr,
                     )
                     migrate(root_directory)
                 except Exception as e:
@@ -124,8 +123,7 @@ def apply_migrations(root_directory):
                     config = _CONFIG_LOADERS[destination](root_directory)
                     config["schema_version"] = destination
                     config.write()
-
-                    print("OK", file=sys.stderr)
+                    _print_err("OK")
     finally:
         try:
             os.unlink(lock.lock_file)

signac/contrib/utility.py

@@ -13,6 +13,10 @@
 logger = logging.getLogger(__name__)
 
 
+def _print_err(*args, **kwargs):
+    print(*args, file=sys.stderr, **kwargs)
+
+
 def _query_yes_no(question, default="yes"):  # pragma: no cover
     """Ask a yes/no question via input() and return their answer.
 

tests/test_find_command_line_interface.py

@@ -2,9 +2,7 @@
 # All rights reserved.
 # This software is licensed under the BSD 3-Clause License.
 import json
-import os
-import sys
-from contextlib import contextmanager
+from contextlib import redirect_stderr
 from io import StringIO
 from itertools import chain
 from json import JSONDecodeError
@@ -64,24 +62,11 @@
 ]
 
 
-@contextmanager
-def redirect_stderr(new_target=None):
-    "Temporarily redirect all output to stderr to new_target."
-    if new_target is None:
-        new_target = StringIO()
-    old_target = sys.stderr
-    try:
-        sys.stderr = new_target
-        yield
-    finally:
-        sys.stderr = old_target
-
-
 class TestFindCommandLineInterface:
     @staticmethod
     def _parse(args):
-        with open(os.devnull, "w") as null:
-            return parse_filter_arg(args, file=null)
+        with redirect_stderr(StringIO()):
+            return parse_filter_arg(args)
 
     def test_interpret_json(self):
         def _assert_equal(q):
@@ -108,6 +93,5 @@ def test_interpret_mixed_key_value(self):
             assert self._parse(["a", json.dumps(expr)]) == {"a": expr}
 
     def test_invalid_json(self):
-        with redirect_stderr():
-            with pytest.raises(JSONDecodeError):
-                parse_filter_arg(['{"x": True}'])
+        with pytest.raises(JSONDecodeError):
+            self._parse(['{"x": True}'])