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
22 from urllib.request import urlopen, urlretrieve
24 PYPI_INSTANCE = "https://pypi.org/pypi"
26 "https://hugovk.github.io/top-pypi-packages/top-pypi-packages-30-days.min.json"
28 INTERNAL_BLACK_REPO = f"{tempfile.gettempdir()}/__black"
30 ArchiveKind = Union[tarfile.TarFile, zipfile.ZipFile]
32 subprocess.run = partial(subprocess.run, check=True) # type: ignore
33 # https://github.com/python/mypy/issues/1484
36 class BlackVersion(NamedTuple):
38 config: Optional[str] = None
41 def get_pypi_download_url(package: str, version: Optional[str]) -> str:
42 with urlopen(PYPI_INSTANCE + f"/{package}/json") as page:
43 metadata = json.load(page)
46 sources = metadata["urls"]
48 if version in metadata["releases"]:
49 sources = metadata["releases"][version]
52 f"No releases found with version ('{version}') tag. "
53 f"Found releases: {metadata['releases'].keys()}"
56 for source in sources:
57 if source["python_version"] == "source":
60 raise ValueError(f"Couldn't find any sources for {package}")
62 return cast(str, source["url"])
65 def get_top_packages() -> List[str]:
66 with urlopen(PYPI_TOP_PACKAGES) as page:
67 result = json.load(page)
69 return [package["project"] for package in result["rows"]]
72 def get_package_source(package: str, version: Optional[str]) -> str:
73 if package == "cpython":
76 return f"https://github.com/python/cpython/archive/{version}.zip"
77 elif package == "pypy":
79 version = "branch/default"
81 f"https://foss.heptapod.net/pypy/pypy/repository/{version}/archive.tar.bz2"
84 return get_pypi_download_url(package, version)
87 def get_archive_manager(local_file: str) -> ArchiveKind:
88 if tarfile.is_tarfile(local_file):
89 return tarfile.open(local_file)
90 elif zipfile.is_zipfile(local_file):
91 return zipfile.ZipFile(local_file)
93 raise ValueError("Unknown archive kind.")
96 def get_first_archive_member(archive: ArchiveKind) -> str:
97 if isinstance(archive, tarfile.TarFile):
98 return archive.getnames()[0]
99 elif isinstance(archive, zipfile.ZipFile):
100 return archive.namelist()[0]
103 def download_and_extract(package: str, version: Optional[str], directory: Path) -> Path:
104 source = get_package_source(package, version)
106 local_file, _ = urlretrieve(source, directory / f"{package}-src")
107 with get_archive_manager(local_file) as archive:
108 archive.extractall(path=directory)
109 result_dir = get_first_archive_member(archive)
110 return directory / result_dir
114 package: str, version: Optional[str], directory: Path
117 return download_and_extract(package, version, directory)
119 print(f"Caught an exception while downloading {package}.")
120 traceback.print_exc()
124 DEFAULT_SLICE = slice(None) # for flake8
127 def download_and_extract_top_packages(
130 limit: slice = DEFAULT_SLICE,
131 ) -> Generator[Path, None, None]:
132 with ThreadPoolExecutor(max_workers=workers) as executor:
133 bound_downloader = partial(get_package, version=None, directory=directory)
134 for package in executor.map(bound_downloader, get_top_packages()[limit]):
135 if package is not None:
139 def git_create_repository(repo: Path) -> None:
140 subprocess.run(["git", "init"], cwd=repo)
141 git_add_and_commit(msg="Initial commit", repo=repo)
144 def git_add_and_commit(msg: str, repo: Path) -> None:
145 subprocess.run(["git", "add", "."], cwd=repo)
146 subprocess.run(["git", "commit", "-m", msg, "--allow-empty"], cwd=repo)
149 def git_switch_branch(
150 branch: str, repo: Path, new: bool = False, from_branch: Optional[str] = None
152 args = ["git", "checkout"]
157 args.append(from_branch)
158 subprocess.run(args, cwd=repo)
161 def init_repos(options: Namespace) -> Tuple[Path, ...]:
162 options.output.mkdir(exist_ok=True)
164 if options.top_packages:
165 source_directories = tuple(
166 download_and_extract_top_packages(
167 directory=options.output,
168 workers=options.workers,
169 limit=slice(None, options.top_packages),
173 source_directories = (
174 download_and_extract(
175 package=options.pypi_package,
176 version=options.version,
177 directory=options.output,
181 for source_directory in source_directories:
182 git_create_repository(source_directory)
184 if options.black_repo is None:
186 ["git", "clone", "https://github.com/psf/black.git", INTERNAL_BLACK_REPO],
189 options.black_repo = options.output / INTERNAL_BLACK_REPO
191 return source_directories
195 def black_runner(version: str, black_repo: Path) -> Path:
196 directory = tempfile.TemporaryDirectory()
197 venv.create(directory.name, with_pip=True)
199 python = Path(directory.name) / "bin" / "python"
200 subprocess.run([python, "-m", "pip", "install", "-e", black_repo])
202 atexit.register(directory.cleanup)
206 def format_repo_with_version(
208 from_branch: Optional[str],
210 black_version: BlackVersion,
211 input_directory: Path,
213 current_branch = f"black-{black_version.version}"
214 git_switch_branch(black_version.version, repo=black_repo)
215 git_switch_branch(current_branch, repo=repo, new=True, from_branch=from_branch)
217 format_cmd: List[Union[Path, str]] = [
218 black_runner(black_version.version, black_repo),
219 (black_repo / "black.py").resolve(),
222 if black_version.config:
223 format_cmd.extend(["--config", input_directory / black_version.config])
225 subprocess.run(format_cmd, cwd=repo, check=False) # ensure the process
226 # continuess to run even it can't format some files. Reporting those
228 git_add_and_commit(f"Format with black:{black_version.version}", repo=repo)
230 return current_branch
233 def format_repos(repos: Tuple[Path, ...], options: Namespace) -> None:
234 black_versions = tuple(
235 BlackVersion(*version.split(":")) for version in options.versions
240 for black_version in black_versions:
241 from_branch = format_repo_with_version(
243 from_branch=from_branch,
244 black_repo=options.black_repo,
245 black_version=black_version,
246 input_directory=options.input,
248 git_switch_branch("main", repo=repo)
250 git_switch_branch("main", repo=options.black_repo)
254 parser = ArgumentParser(
255 description="""Black Gallery is a script that
256 automates the process of applying different Black versions to a selected
257 PyPI package and seeing the results between versions."""
260 group = parser.add_mutually_exclusive_group(required=True)
261 group.add_argument("-p", "--pypi-package", help="PyPI package to download.")
263 "-t", "--top-packages", help="Top n PyPI packages to download.", type=int
266 parser.add_argument("-b", "--black-repo", help="Black's Git repository.", type=Path)
271 "Version for given PyPI package. Will be discarded if used with -t option."
278 "Maximum number of threads to download with at the same time. "
279 "Will be discarded if used with -p option."
285 default=Path("/input"),
287 help="Input directory to read configuration.",
292 default=Path("/output"),
294 help="Output directory to download and put result artifacts.",
296 parser.add_argument("versions", nargs="*", default=("main",), help="")
298 options = parser.parse_args()
299 repos = init_repos(options)
300 format_repos(repos, options)
303 if __name__ == "__main__":