]> git.madduck.net Git - etc/vim.git/blob - src/black_primer/cli.py

madduck's git repository

Every one of the projects in this repository is available at the canonical URL git://git.madduck.net/madduck/pub/<projectpath> — see each project's metadata for the exact URL.

All patches and comments are welcome. Please squash your changes to logical commits before using git-format-patch and git-send-email to patches@git.madduck.net. If you'd read over the Git project's submission guidelines and adhered to them, I'd be especially grateful.

SSH access, as well as push access can be individually arranged.

If you use my repositories frequently, consider adding the following snippet to ~/.gitconfig and using the third clone URL listed for each project:

[url "git://git.madduck.net/madduck/"]
  insteadOf = madduck:

fix error message for match (#2649)
[etc/vim.git] / 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()