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     no_diff: bool,
  43     rebase: bool,
  44     workdir: str,
  45     workers: int,
  46 ) -> int:
  47     work_path = Path(workdir)
  48     if not work_path.exists():
  49         LOG.debug(f"Creating {work_path}")
  50         work_path.mkdir()
  51
  52     if not which("black"):
  53         LOG.error("Can not find 'black' executable in PATH. No point in running")
  54         return -1
  55
  56     try:
  57         ret_val = await lib.process_queue(
  58             config,
  59             work_path,
  60             workers,
  61             keep,
  62             long_checkouts,
  63             rebase,
  64             no_diff,
  65         )
  66         return int(ret_val)
  67     finally:
  68         if not keep and work_path.exists():
  69             LOG.debug(f"Removing {work_path}")
  70             rmtree(work_path, onerror=lib.handle_PermissionError)
  71
  72     return -2
  73
  74
  75 @click.command(context_settings={"help_option_names": ["-h", "--help"]})
  76 @click.option(
  77     "-c",
  78     "--config",
  79     default=str(DEFAULT_CONFIG),
  80     type=click.Path(exists=True),
  81     show_default=True,
  82     help="JSON config file path",
  83 )
  84 @click.option(
  85     "--debug",
  86     is_flag=True,
  87     callback=_handle_debug,
  88     show_default=True,
  89     help="Turn on debug logging",
  90 )
  91 @click.option(
  92     "-k",
  93     "--keep",
  94     is_flag=True,
  95     show_default=True,
  96     help="Keep workdir + repos post run",
  97 )
  98 @click.option(
  99     "-L",
 100     "--long-checkouts",
 101     is_flag=True,
 102     show_default=True,
 103     help="Pull big projects to test",
 104 )
 105 @click.option(
 106     "--no-diff",
 107     is_flag=True,
 108     show_default=True,
 109     help="Disable showing source file changes in black output",
 110 )
 111 @click.option(
 112     "-R",
 113     "--rebase",
 114     is_flag=True,
 115     show_default=True,
 116     help="Rebase project if already checked out",
 117 )
 118 @click.option(
 119     "-w",
 120     "--workdir",
 121     default=str(DEFAULT_WORKDIR),
 122     type=click.Path(exists=False),
 123     show_default=True,
 124     help="Directory path for repo checkouts",
 125 )
 126 @click.option(
 127     "-W",
 128     "--workers",
 129     default=2,
 130     type=int,
 131     show_default=True,
 132     help="Number of parallel worker coroutines",
 133 )
 134 @click.pass_context
 135 def main(ctx: click.core.Context, **kwargs: Any) -> None:
 136     """primer - prime projects for blackening... 🏴"""
 137     LOG.debug(f"Starting {sys.argv[0]}")
 138     # TODO: Change to asyncio.run when Black >= 3.7 only
 139     loop = asyncio.get_event_loop()
 140     try:
 141         ctx.exit(loop.run_until_complete(async_main(**kwargs)))
 142     finally:
 143         loop.close()
 144
 145
 146 if __name__ == "__main__":  # pragma: nocover
 147     main()