src/black_primer/cli.py

   1 # coding=utf8
   2
   3 import asyncio
   4 import logging
   5 import sys
   6 from datetime import datetime
   7 from pathlib import Path
   8 from shutil import rmtree, which
   9 from tempfile import gettempdir
  10 from typing import Any, Union
  11
  12 import click
  13
  14 from black_primer import lib
  15
  16
  17 DEFAULT_CONFIG = Path(__file__).parent / "primer.json"
  18 _timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
  19 DEFAULT_WORKDIR = Path(gettempdir()) / f"primer.{_timestamp}"
  20 LOG = logging.getLogger(__name__)
  21
  22
  23 def _handle_debug(
  24     ctx: click.core.Context,
  25     param: Union[click.core.Option, click.core.Parameter],
  26     debug: Union[bool, int, str],
  27 ) -> Union[bool, int, str]:
  28     """Turn on debugging if asked otherwise INFO default"""
  29     log_level = logging.DEBUG if debug else logging.INFO
  30     logging.basicConfig(
  31         format="[%(asctime)s] %(levelname)s: %(message)s (%(filename)s:%(lineno)d)",
  32         level=log_level,
  33     )
  34     return debug
  35
  36
  37 async def async_main(
  38     config: str,
  39     debug: bool,
  40     keep: bool,
  41     long_checkouts: bool,
  42     rebase: bool,
  43     workdir: str,
  44     workers: int,
  45 ) -> int:
  46     work_path = Path(workdir)
  47     if not work_path.exists():
  48         LOG.debug(f"Creating {work_path}")
  49         work_path.mkdir()
  50
  51     if not which("black"):
  52         LOG.error("Can not find 'black' executable in PATH. No point in running")
  53         return -1
  54
  55     try:
  56         ret_val = await lib.process_queue(
  57             config, work_path, workers, keep, long_checkouts, rebase
  58         )
  59         return int(ret_val)
  60     finally:
  61         if not keep and work_path.exists():
  62             LOG.debug(f"Removing {work_path}")
  63             rmtree(work_path, onerror=lib.handle_PermissionError)
  64
  65     return -2
  66
  67
  68 @click.command(context_settings={"help_option_names": ["-h", "--help"]})
  69 @click.option(
  70     "-c",
  71     "--config",
  72     default=str(DEFAULT_CONFIG),
  73     type=click.Path(exists=True),
  74     show_default=True,
  75     help="JSON config file path",
  76 )
  77 @click.option(
  78     "--debug",
  79     is_flag=True,
  80     callback=_handle_debug,
  81     show_default=True,
  82     help="Turn on debug logging",
  83 )
  84 @click.option(
  85     "-k",
  86     "--keep",
  87     is_flag=True,
  88     show_default=True,
  89     help="Keep workdir + repos post run",
  90 )
  91 @click.option(
  92     "-L",
  93     "--long-checkouts",
  94     is_flag=True,
  95     show_default=True,
  96     help="Pull big projects to test",
  97 )
  98 @click.option(
  99     "-R",
 100     "--rebase",
 101     is_flag=True,
 102     show_default=True,
 103     help="Rebase project if already checked out",
 104 )
 105 @click.option(
 106     "-w",
 107     "--workdir",
 108     default=str(DEFAULT_WORKDIR),
 109     type=click.Path(exists=False),
 110     show_default=True,
 111     help="Directory path for repo checkouts",
 112 )
 113 @click.option(
 114     "-W",
 115     "--workers",
 116     default=2,
 117     type=int,
 118     show_default=True,
 119     help="Number of parallel worker coroutines",
 120 )
 121 @click.pass_context
 122 def main(ctx: click.core.Context, **kwargs: Any) -> None:
 123     """primer - prime projects for blackening... 🏴"""
 124     LOG.debug(f"Starting {sys.argv[0]}")
 125     # TODO: Change to asyncio.run when Black >= 3.7 only
 126     loop = asyncio.get_event_loop()
 127     try:
 128         ctx.exit(loop.run_until_complete(async_main(**kwargs)))
 129     finally:
 130         loop.close()
 131
 132
 133 if __name__ == "__main__":  # pragma: nocover
 134     main()