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", "-m", 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", "-m", 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", "-m", 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", "-m", 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: