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
 108     finally:
 109         if not keep and work_path.exists():
 110             LOG.debug(f"Removing {work_path}")
 111             rmtree(work_path, onerror=lib.handle_PermissionError)
 112
 113
 114 @click.command(context_settings={"help_option_names": ["-h", "--help"]})
 115 @click.option(
 116     "-c",
 117     "--config",
 118     default=str(DEFAULT_CONFIG),
 119     type=click.Path(exists=True),
 120     show_default=True,
 121     help="JSON config file path",
 122     # Eager - because config path is used by other callback options
 123     is_eager=True,
 124 )
 125 @click.option(
 126     "--debug",
 127     is_flag=True,
 128     callback=_handle_debug,
 129     show_default=True,
 130     help="Turn on debug logging",
 131 )
 132 @click.option(
 133     "-k",
 134     "--keep",
 135     is_flag=True,
 136     show_default=True,
 137     help="Keep workdir + repos post run",
 138 )
 139 @click.option(
 140     "-L",
 141     "--long-checkouts",
 142     is_flag=True,
 143     show_default=True,
 144     help="Pull big projects to test",
 145 )
 146 @click.option(
 147     "--no-diff",
 148     is_flag=True,
 149     show_default=True,
 150     help="Disable showing source file changes in black output",
 151 )
 152 @click.option(
 153     "--projects",
 154     default=",".join(DEFAULT_PROJECTS),
 155     callback=_projects_callback,
 156     show_default=True,
 157     help="Comma separated list of projects to run",
 158 )
 159 @click.option(
 160     "-R",
 161     "--rebase",
 162     is_flag=True,
 163     show_default=True,
 164     help="Rebase project if already checked out",
 165 )
 166 @click.option(
 167     "-w",
 168     "--workdir",
 169     default=str(DEFAULT_WORKDIR),
 170     type=click.Path(exists=False),
 171     show_default=True,
 172     help="Directory path for repo checkouts",
 173 )
 174 @click.option(
 175     "-W",
 176     "--workers",
 177     default=2,
 178     type=int,
 179     show_default=True,
 180     help="Number of parallel worker coroutines",
 181 )
 182 @click.pass_context
 183 def main(ctx: click.core.Context, **kwargs: Any) -> None:
 184     """primer - prime projects for blackening... 🏴"""
 185     LOG.debug(f"Starting {sys.argv[0]}")
 186     # TODO: Change to asyncio.run when Black >= 3.7 only
 187     loop = asyncio.get_event_loop()
 188     try:
 189         ctx.exit(loop.run_until_complete(async_main(**kwargs)))
 190     finally:
 191         loop.close()
 192
 193
 194 if __name__ == "__main__":  # pragma: nocover
 195     main()