Add primer CI tool 🏴

setup.py

@@ -62,7 +62,7 @@ def get_long_description() -> str:
     license="MIT",
     py_modules=["_black_version"],
     ext_modules=ext_modules,
-    packages=["blackd", "black", "blib2to3", "blib2to3.pgen2"],
+    packages=["blackd", "black", "blib2to3", "blib2to3.pgen2", "black_primer"],
     package_dir={"": "src"},
     package_data={"blib2to3": ["*.txt"], "black": ["py.typed"]},
     python_requires=">=3.6",
@@ -102,6 +102,7 @@ def get_long_description() -> str:
         "console_scripts": [
             "black=black:patched_main",
             "blackd=blackd:patched_main [d]",
+            "black-primer=black_primer.cli:main",
         ]
     },
 )

src/black_primer/cli.py

@@ -0,0 +1,135 @@
+#!/usr/bin/env python3
+
+import asyncio
+import logging
+import sys
+from datetime import datetime
+from os import cpu_count
+from pathlib import Path
+from shutil import rmtree, which
+from tempfile import gettempdir
+from typing import Any, Union
+
+import click
+
+from black_primer import lib
+
+
+DEFAULT_CONFIG = Path(__file__).parent / "primer.json"
+_timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
+DEFAULT_WORKDIR = Path(gettempdir()) / f"primer.{_timestamp}"
+LOG = logging.getLogger(__name__)
+
+
+def _handle_debug(
+    ctx: click.core.Context,
+    param: Union[click.core.Option, click.core.Parameter],
+    debug: Union[bool, int, str],
+) -> Union[bool, int, str]:
+    """Turn on debugging if asked otherwise INFO default"""
+    log_level = logging.DEBUG if debug else logging.INFO
+    logging.basicConfig(
+        format="[%(asctime)s] %(levelname)s: %(message)s (%(filename)s:%(lineno)d)",
+        level=log_level,
+    )
+    return debug
+
+
+async def async_main(
+    config: str,
+    debug: bool,
+    keep: bool,
+    long_checkouts: bool,
+    rebase: bool,
+    workdir: str,
+    workers: int,
+) -> int:
+    work_path = Path(workdir)
+    if not work_path.exists():
+        LOG.debug(f"Creating {work_path}")
+        work_path.mkdir()
+
+    if not which("black"):
+        LOG.error(f"Can not find 'black' executable in PATH. No point in running")
+        return -1
+
+    try:
+        ret_val = await lib.process_queue(
+            config, work_path, workers, keep, long_checkouts, rebase
+        )
+        return int(ret_val)
+    finally:
+        if not keep and work_path.exists():
+            LOG.debug(f"Removing {work_path}")
+            rmtree(work_path)
+
+    return -1
+
+
+@click.command(context_settings={"help_option_names": ["-h", "--help"]})
+@click.option(
+    "-c",
+    "--config",
+    default=str(DEFAULT_CONFIG),
+    type=click.Path(exists=True),
+    show_default=True,
+    help="JSON config file path",
+)
+@click.option(
+    "--debug",
+    is_flag=True,
+    callback=_handle_debug,
+    show_default=True,
+    help="Turn on debug logging",
+)
+@click.option(
+    "-k",
+    "--keep",
+    is_flag=True,
+    show_default=True,
+    help="Keep workdir + repos post run",
+)
+@click.option(
+    "-L",
+    "--long-checkouts",
+    is_flag=True,
+    show_default=True,
+    help="Pull big projects to test",
+)
+@click.option(
+    "-R",
+    "--rebase",
+    is_flag=True,
+    show_default=True,
+    help="Rebase project if already checked out",
+)
+@click.option(
+    "-w",
+    "--workdir",
+    default=str(DEFAULT_WORKDIR),
+    type=click.Path(exists=False),
+    show_default=True,
+    help="Directory Path for repo checkouts",
+)
+@click.option(
+    "-W",
+    "--workers",
+    default=int((cpu_count() or 4) / 2) or 1,
+    type=int,
+    show_default=True,
+    help="Number of parallel worker coroutines",
+)
+@click.pass_context
+def main(ctx: click.core.Context, **kwargs: Any) -> None:
+    """primer - prime projects for blackening ... 🏴"""
+    LOG.debug(f"Starting {sys.argv[0]}")
+    # TODO: Change to asyncio.run when black >= 3.7 only
+    loop = asyncio.get_event_loop()
+    try:
+        ctx.exit(loop.run_until_complete(async_main(**kwargs)))
+    finally:
+        loop.close()
+
+
+if __name__ == "__main__":
+    main()