src/black_primer/cli.py

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