src/black_primer/cli.py

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