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