feat: new cli

pyaptly/cli.py

@@ -1,24 +1,53 @@
 """python-click based command line interface for pyaptly."""
 
+import logging
 import sys
 from pathlib import Path
+from subprocess import CalledProcessError
 
 import click
 
-# I decided it is a good pattern to do lazy imports in the cli module. I had to
-# do this in a few other CLIs for startup performance.
+lg = logging.getLogger(__name__)
 
 
-# TODO this makes the legacy command more usable. remove and set the entry point
-# back to `pyaptly = 'pyaptly.cli:cli'
 def entry_point():
     """Fix args then call click."""
+    # TODO this makes the legacy command more usable. remove legacy commands when
+    # we are out of beta
     argv = list(sys.argv)
     len_argv = len(argv)
     if len_argv > 0 and argv[0].endswith("pyaptly"):
         if len_argv > 2 and argv[1] == "legacy" and argv[2] != "--":
             argv = argv[:2] + ["--"] + argv[2:]
-    cli.main(argv[1:])
+
+    try:
+        cli.main(argv[1:])
+    except CalledProcessError:
+        pass  # already logged
+    except Exception as e:
+        from . import util
+
+        path = util.write_traceback()
+        tb = f"Wrote traceback to: {path}"
+        msg = " ".join([str(x) for x in e.args])
+        lg.error(f"{msg}\n               {tb}")
+
+
+# I want to release the new cli interface with 2.0, so we do not repeat breaking changes.
+# But changing all functions that use argparse, means also changing all the tests, which
+# (ab)use the argparse interface. So we currently fake that interface, so we can roll-out
+# the new interface early.
+# TODO: remove this, once argparse is gone
+class FakeArgs:
+    """Helper for compatiblity."""
+
+    def __init__(self, **kwargs):
+        for key, value in kwargs.items():
+            setattr(self, key, value)
+
+
+# I decided it is a good pattern to do lazy imports in the cli module. I had to
+# do this in a few other CLIs for startup performance.
 
 
 @click.group()
@@ -47,6 +76,98 @@ def legacy(passthrough):
     main.main(argv=passthrough)
 
 
+@cli.command()
+@click.option("--info/--no-info", "-i/-ni", default=False, type=bool)
+@click.option("--debug/--no-debug", "-d/-nd", default=False, type=bool)
+@click.option(
+    "--pretend/--no-pretend",
+    "-p/-np",
+    default=False,
+    type=bool,
+    help="Do not change anything",
+)
+@click.argument("config", type=click.Path(file_okay=True, dir_okay=False, exists=True))
+@click.argument("task", type=click.Choice(["create"]))
+@click.option("--repo-name", "-n", default="all", type=str, help='deafult: "all"')
+def repo(**kwargs):
+    """Create aptly repos."""
+    from . import main, repo
+
+    fake_args = FakeArgs(**kwargs)
+    main.setup_logger(fake_args)
+    cfg = main.prepare(fake_args)
+    repo.repo(cfg, args=fake_args)
+
+
+@cli.command()
+@click.option("--info/--no-info", "-i/-ni", default=False, type=bool)
+@click.option("--debug/--no-debug", "-d/-nd", default=False, type=bool)
+@click.option(
+    "--pretend/--no-pretend",
+    "-p/-np",
+    default=False,
+    type=bool,
+    help="Do not change anything",
+)
+@click.argument("config", type=click.Path(file_okay=True, dir_okay=False, exists=True))
+@click.argument("task", type=click.Choice(["create", "update"]))
+@click.option("--mirror-name", "-n", default="all", type=str, help='deafult: "all"')
+def mirror(**kwargs):
+    """Manage aptly mirrors."""
+    from . import main, mirror
+
+    fake_args = FakeArgs(**kwargs)
+    main.setup_logger(fake_args)
+    cfg = main.prepare(fake_args)
+    mirror.mirror(cfg, args=fake_args)
+
+
+@cli.command()
+@click.option("--info/--no-info", "-i/-ni", default=False, type=bool)
+@click.option("--debug/--no-debug", "-d/-nd", default=False, type=bool)
+@click.option(
+    "--pretend/--no-pretend",
+    "-p/-np",
+    default=False,
+    type=bool,
+    help="Do not change anything",
+)
+@click.argument("config", type=click.Path(file_okay=True, dir_okay=False, exists=True))
+@click.argument("task", type=click.Choice(["create", "update"]))
+@click.option("--snapshot-name", "-n", default="all", type=str, help='deafult: "all"')
+def snapshot(**kwargs):
+    """Manage aptly snapshots."""
+    from . import main, snapshot
+
+    fake_args = FakeArgs(**kwargs)
+    main.setup_logger(fake_args)
+    cfg = main.prepare(fake_args)
+    snapshot.snapshot(cfg, args=fake_args)
+
+
+@cli.command()
+@click.option("--info/--no-info", "-i/-ni", default=False, type=bool)
+@click.option("--debug/--no-debug", "-d/-nd", default=False, type=bool)
+@click.option(
+    "--pretend/--no-pretend",
+    "-p/-np",
+    default=False,
+    type=bool,
+    help="Do not change anything",
+)
+@click.argument("config", type=click.Path(file_okay=True, dir_okay=False, exists=True))
+@click.argument("task", type=click.Choice(["create", "update"]))
+@click.option("--publish-name", "-n", default="all", type=str, help='deafult: "all"')
+def publish(**kwargs):
+    """Manage aptly publishs."""
+    from . import main, publish
+
+    fake_args = FakeArgs(**kwargs)
+    main.setup_logger(fake_args)
+    cfg = main.prepare(fake_args)
+    publish.publish(cfg, args=fake_args)
+
+
 @cli.command(help="convert yaml- to toml-comfig")
 @click.argument(
     "yaml_path",

pyaptly/main.py

@@ -22,13 +22,42 @@
 lg = logging.getLogger(__name__)
 
 
+def setup_logger(args):
+    """Setup the logger."""
+    global _logging_setup
+    root = logging.getLogger()
+    formatter = custom_logger.CustomFormatter()
+    if not _logging_setup:  # noqa
+        handler = logging.StreamHandler(sys.stderr)
+        handler.setFormatter(formatter)
+        root.addHandler(handler)
+        root.setLevel(logging.WARNING)
+        handler.setLevel(logging.WARNING)
+        if args.info:
+            root.setLevel(logging.INFO)
+            handler.setLevel(logging.INFO)
+        if args.debug:
+            root.setLevel(logging.DEBUG)
+            handler.setLevel(logging.DEBUG)
+        _logging_setup = True  # noqa
+
+
+def prepare(args):
+    """Set pretend mode, read config and load state."""
+    command.Command.pretend_mode = args.pretend
+
+    with open(args.config, "rb") as f:
+        cfg = tomli.load(f)
+    state_reader.state.read()
+    return cfg
+
+
 def main(argv=None):
     """Define parsers and executes commands.
 
     :param argv: Arguments usually taken from sys.argv
     :type  argv: list
     """
-    global _logging_setup
     if not argv:  # pragma: no cover
         argv = sys.argv[1:]
     parser = argparse.ArgumentParser(description="Manage aptly")
@@ -78,31 +107,8 @@ def main(argv=None):
     repo_parser.add_argument("repo_name", type=str, nargs="?", default="all")
 
     args = parser.parse_args(argv)
-    root = logging.getLogger()
-    formatter = custom_logger.CustomFormatter()
-    if not _logging_setup:  # noqa
-        handler = logging.StreamHandler(sys.stderr)
-        handler.setFormatter(formatter)
-        root.addHandler(handler)
-        root.setLevel(logging.WARNING)
-        handler.setLevel(logging.WARNING)
-        if args.info:
-            root.setLevel(logging.INFO)
-            handler.setLevel(logging.INFO)
-        if args.debug:
-            root.setLevel(logging.DEBUG)
-            handler.setLevel(logging.DEBUG)
-    if args.pretend:
-        command.Command.pretend_mode = True
-    else:
-        command.Command.pretend_mode = False
-
-        _logging_setup = True  # noqa
-    lg.debug("Args: %s", vars(args))
-
-    with open(args.config, "rb") as f:
-        cfg = tomli.load(f)
-    state_reader.state.read()
+    setup_logger(args)
+    cfg = prepare(args)
 
     # run function for selected subparser
     args.func(cfg, args)

pyaptly/util.py

@@ -3,12 +3,14 @@
 import logging
 import os
 import subprocess
+import traceback
 from pathlib import Path
-
-from colorama import Fore, init
 from subprocess import PIPE, CalledProcessError  # noqa: F401
+from tempfile import NamedTemporaryFile
 from typing import Optional, Sequence
 
+from colorama import Fore, init
+
 _DEFAULT_KEYSERVER: str = "hkps://keys.openpgp.org"
 _PYTEST_KEYSERVER: Optional[str] = None
 
@@ -28,6 +30,13 @@
 lg = logging.getLogger(__name__)
 
 
+def write_traceback():  # pragma: no cover
+    with NamedTemporaryFile("w", delete=False) as tmp:
+        tmp.write(traceback.format_exc())
+        tmp.close()
+    return tmp.name
+
+
 def isatty():
     global _isatty_cache
     if _isatty_cache is None: