src/black_primer/cli.py

   1 # coding=utf8
   2
   3 import asyncio
   4 import json
   5 import logging
   6 import sys
   7 from datetime import datetime
   8 from pathlib import Path
   9 from shutil import rmtree, which
  10 from tempfile import gettempdir
  11 from typing import Any, List, Optional, Union
  12
  13 import click
  14
  15 from black_primer import lib
  16
  17 # If our environment has uvloop installed lets use it
  18 try:
  19     import uvloop
  20
  21     uvloop.install()
  22 except ImportError:
  23     pass
  24
  25
  26 DEFAULT_CONFIG = Path(__file__).parent / "primer.json"
  27 _timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
  28 DEFAULT_WORKDIR = Path(gettempdir()) / f"primer.{_timestamp}"
  29 LOG = logging.getLogger(__name__)
  30
  31
  32 def _handle_debug(
  33     ctx: Optional[click.core.Context],
  34     param: Optional[Union[click.core.Option, click.core.Parameter]],
  35     debug: Union[bool, int, str],
  36 ) -> Union[bool, int, str]:
  37     """Turn on debugging if asked otherwise INFO default"""
  38     log_level = logging.DEBUG if debug else logging.INFO
  39     logging.basicConfig(
  40         format="[%(asctime)s] %(levelname)s: %(message)s (%(filename)s:%(lineno)d)",
  41         level=log_level,
  42     )
  43     return debug
  44
  45
  46 def load_projects(config_path: Path) -> List[str]:
  47     with open(config_path) as config:
  48         return sorted(json.load(config)["projects"].keys())
  49
  50
  51 # Unfortunately does import time file IO - but appears to be the only
  52 # way to get `black-primer --help` to show projects list
  53 DEFAULT_PROJECTS = load_projects(DEFAULT_CONFIG)
  54
  55
  56 def _projects_callback(
  57     ctx: click.core.Context,
  58     param: Optional[Union[click.core.Option, click.core.Parameter]],
  59     projects: str,
  60 ) -> List[str]:
  61     requested_projects = set(projects.split(","))
  62     available_projects = set(
  63         DEFAULT_PROJECTS
  64         if str(DEFAULT_CONFIG) == ctx.params["config"]
  65         else load_projects(ctx.params["config"])
  66     )
  67
  68     unavailable = requested_projects - available_projects
  69     if unavailable:
  70         LOG.error(f"Projects not found: {unavailable}. Available: {available_projects}")
  71
  72     return sorted(requested_projects & available_projects)
  73
  74
  75 async def async_main(
  76     config: str,
  77     debug: bool,
  78     keep: bool,
  79     long_checkouts: bool,
  80     no_diff: bool,
  81     projects: List[str],
  82     rebase: bool,
  83     workdir: str,
  84     workers: int,
  85 ) -> int:
  86     work_path = Path(workdir)
  87     if not work_path.exists():
  88         LOG.debug(f"Creating {work_path}")
  89         work_path.mkdir()
  90
  91     if not which("black"):
  92         LOG.error("Can not find 'black' executable in PATH. No point in running")
  93         return -1
  94
  95     try:
  96         ret_val = await lib.process_queue(
  97             config,
  98             work_path,
  99             workers,
 100             projects,
 101             keep,
 102             long_checkouts,
 103             rebase,
 104             no_diff,
 105         )
 106         return int(ret_val)
 107     finally:
 108         if not keep and work_path.exists():
 109             LOG.debug(f"Removing {work_path}")
 110             rmtree(work_path, onerror=lib.handle_PermissionError)
 111
 112     return -2
 113
 114
 115 @click.command(context_settings={"help_option_names": ["-h", "--help"]})
 116 @click.option(
 117     "-c",
 118     "--config",
 119     default=str(DEFAULT_CONFIG),
 120     type=click.Path(exists=True),
 121     show_default=True,
 122     help="JSON config file path",
 123     # Eager - because config path is used by other callback options
 124     is_eager=True,
 125 )
 126 @click.option(
 127     "--debug",
 128     is_flag=True,
 129     callback=_handle_debug,
 130     show_default=True,
 131     help="Turn on debug logging",
 132 )
 133 @click.option(
 134     "-k",
 135     "--keep",
 136     is_flag=True,
 137     show_default=True,
 138     help="Keep workdir + repos post run",
 139 )
 140 @click.option(
 141     "-L",
 142     "--long-checkouts",
 143     is_flag=True,
 144     show_default=True,
 145     help="Pull big projects to test",
 146 )
 147 @click.option(
 148     "--no-diff",
 149     is_flag=True,
 150     show_default=True,
 151     help="Disable showing source file changes in black output",
 152 )
 153 @click.option(
 154     "--projects",
 155     default=",".join(DEFAULT_PROJECTS),
 156     callback=_projects_callback,
 157     show_default=True,
 158     help="Comma separated list of projects to run",
 159 )
 160 @click.option(
 161     "-R",
 162     "--rebase",
 163     is_flag=True,
 164     show_default=True,
 165     help="Rebase project if already checked out",
 166 )
 167 @click.option(
 168     "-w",
 169     "--workdir",
 170     default=str(DEFAULT_WORKDIR),
 171     type=click.Path(exists=False),
 172     show_default=True,
 173     help="Directory path for repo checkouts",
 174 )
 175 @click.option(
 176     "-W",
 177     "--workers",
 178     default=2,
 179     type=int,
 180     show_default=True,
 181     help="Number of parallel worker coroutines",
 182 )
 183 @click.pass_context
 184 def main(ctx: click.core.Context, **kwargs: Any) -> None:
 185     """primer - prime projects for blackening... 🏴"""
 186     LOG.debug(f"Starting {sys.argv[0]}")
 187     # TODO: Change to asyncio.run when Black >= 3.7 only
 188     loop = asyncio.get_event_loop()
 189     try:
 190         ctx.exit(loop.run_until_complete(async_main(**kwargs)))
 191     finally:
 192         loop.close()
 193
 194
 195 if __name__ == "__main__":  # pragma: nocover
 196     main()