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.
9 from argparse import ArgumentParser, Namespace
10 from concurrent.futures import ThreadPoolExecutor
11 from functools import lru_cache, partial
12 from pathlib import Path
13 from typing import ( # type: ignore # typing can't see Literal
23 from urllib.request import urlopen, urlretrieve
25 PYPI_INSTANCE = "https://pypi.org/pypi"
27 "https://hugovk.github.io/top-pypi-packages/top-pypi-packages-{days}-days.json"
29 INTERNAL_BLACK_REPO = f"{tempfile.gettempdir()}/__black"
31 ArchiveKind = Union[tarfile.TarFile, zipfile.ZipFile]
32 Days = Union[Literal[30], Literal[365]]
34 subprocess.run = partial(subprocess.run, check=True) # type: ignore
35 # https://github.com/python/mypy/issues/1484
38 class BlackVersion(NamedTuple):
40 config: Optional[str] = None
43 def get_pypi_download_url(package: str, version: Optional[str]) -> str:
44 with urlopen(PYPI_INSTANCE + f"/{package}/json") as page:
45 metadata = json.load(page)
48 sources = metadata["urls"]
50 if version in metadata["releases"]:
51 sources = metadata["releases"][version]
54 f"No releases found with version ('{version}') tag. "
55 f"Found releases: {metadata['releases'].keys()}"
58 for source in sources:
59 if source["python_version"] == "source":
62 raise ValueError(f"Couldn't find any sources for {package}")
64 return cast(str, source["url"])
67 def get_top_packages(days: Days) -> List[str]:
68 with urlopen(PYPI_TOP_PACKAGES.format(days=days)) as page:
69 result = json.load(page)
71 return [package["project"] for package in result["rows"]]
74 def get_package_source(package: str, version: Optional[str]) -> str:
75 if package == "cpython":
78 return f"https://github.com/python/cpython/archive/{version}.zip"
79 elif package == "pypy":
81 version = "branch/default"
83 f"https://foss.heptapod.net/pypy/pypy/repository/{version}/archive.tar.bz2"
86 return get_pypi_download_url(package, version)
89 def get_archive_manager(local_file: str) -> ArchiveKind:
90 if tarfile.is_tarfile(local_file):
91 return tarfile.open(local_file)
92 elif zipfile.is_zipfile(local_file):
93 return zipfile.ZipFile(local_file)
95 raise ValueError("Unknown archive kind.")
98 def get_first_archive_member(archive: ArchiveKind) -> str:
99 if isinstance(archive, tarfile.TarFile):
100 return archive.getnames()[0]
101 elif isinstance(archive, zipfile.ZipFile):
102 return archive.namelist()[0]
105 def download_and_extract(package: str, version: Optional[str], directory: Path) -> Path:
106 source = get_package_source(package, version)
108 local_file, _ = urlretrieve(source, directory / f"{package}-src")
109 with get_archive_manager(local_file) as archive:
110 archive.extractall(path=directory)
111 result_dir = get_first_archive_member(archive)
112 return directory / result_dir
116 package: str, version: Optional[str], directory: Path
119 return download_and_extract(package, version, directory)
121 print(f"Caught an exception while downloading {package}.")
122 traceback.print_exc()
126 DEFAULT_SLICE = slice(None) # for flake8
129 def download_and_extract_top_packages(
133 limit: slice = DEFAULT_SLICE,
134 ) -> Generator[Path, None, None]:
135 with ThreadPoolExecutor(max_workers=workers) as executor:
136 bound_downloader = partial(get_package, version=None, directory=directory)
137 for package in executor.map(bound_downloader, get_top_packages(days)[limit]):
138 if package is not None:
142 def git_create_repository(repo: Path) -> None:
143 subprocess.run(["git", "init"], cwd=repo)
144 git_add_and_commit(msg="Initial commit", repo=repo)
147 def git_add_and_commit(msg: str, repo: Path) -> None:
148 subprocess.run(["git", "add", "."], cwd=repo)
149 subprocess.run(["git", "commit", "-m", msg, "--allow-empty"], cwd=repo)
152 def git_switch_branch(
153 branch: str, repo: Path, new: bool = False, from_branch: Optional[str] = None
155 args = ["git", "checkout"]
160 args.append(from_branch)
161 subprocess.run(args, cwd=repo)
164 def init_repos(options: Namespace) -> Tuple[Path, ...]:
165 options.output.mkdir(exist_ok=True)
167 if options.top_packages:
168 source_directories = tuple(
169 download_and_extract_top_packages(
170 directory=options.output,
171 workers=options.workers,
172 limit=slice(None, options.top_packages),
176 source_directories = (
177 download_and_extract(
178 package=options.pypi_package,
179 version=options.version,
180 directory=options.output,
184 for source_directory in source_directories:
185 git_create_repository(source_directory)
187 if options.black_repo is None:
189 ["git", "clone", "https://github.com/psf/black.git", INTERNAL_BLACK_REPO],
192 options.black_repo = options.output / INTERNAL_BLACK_REPO
194 return source_directories
198 def black_runner(version: str, black_repo: Path) -> Path:
199 directory = tempfile.TemporaryDirectory()
200 venv.create(directory.name, with_pip=True)
202 python = Path(directory.name) / "bin" / "python"
203 subprocess.run([python, "-m", "pip", "install", "-e", black_repo])
205 atexit.register(directory.cleanup)
209 def format_repo_with_version(
211 from_branch: Optional[str],
213 black_version: BlackVersion,
214 input_directory: Path,
216 current_branch = f"black-{black_version.version}"
217 git_switch_branch(black_version.version, repo=black_repo)
218 git_switch_branch(current_branch, repo=repo, new=True, from_branch=from_branch)
220 format_cmd: List[Union[Path, str]] = [
221 black_runner(black_version.version, black_repo),
222 (black_repo / "black.py").resolve(),
225 if black_version.config:
226 format_cmd.extend(["--config", input_directory / black_version.config])
228 subprocess.run(format_cmd, cwd=repo, check=False) # ensure the process
229 # continuess to run even it can't format some files. Reporting those
231 git_add_and_commit(f"Format with black:{black_version.version}", repo=repo)
233 return current_branch
236 def format_repos(repos: Tuple[Path, ...], options: Namespace) -> None:
237 black_versions = tuple(
238 BlackVersion(*version.split(":")) for version in options.versions
243 for black_version in black_versions:
244 from_branch = format_repo_with_version(
246 from_branch=from_branch,
247 black_repo=options.black_repo,
248 black_version=black_version,
249 input_directory=options.input,
251 git_switch_branch("main", repo=repo)
253 git_switch_branch("main", repo=options.black_repo)
257 parser = ArgumentParser(
258 description="""Black Gallery is a script that
259 automates the process of applying different Black versions to a selected
260 PyPI package and seeing the results between versions."""
263 group = parser.add_mutually_exclusive_group(required=True)
264 group.add_argument("-p", "--pypi-package", help="PyPI package to download.")
266 "-t", "--top-packages", help="Top n PyPI packages to download.", type=int
269 parser.add_argument("-b", "--black-repo", help="Black's Git repository.", type=Path)
274 "Version for given PyPI package. Will be discarded if used with -t option."
281 "Maximum number of threads to download with at the same time. "
282 "Will be discarded if used with -p option."
288 default=Path("/input"),
290 help="Input directory to read configuration.",
295 default=Path("/output"),
297 help="Output directory to download and put result artifacts.",
299 parser.add_argument("versions", nargs="*", default=("main",), help="")
301 options = parser.parse_args()
302 repos = init_repos(options)
303 format_repos(repos, options)
306 if __name__ == "__main__":