|
| 1 | +import os |
| 2 | +import json |
| 3 | +import itertools |
| 4 | +from typing import Any |
| 5 | +from types import MappingProxyType |
| 6 | +from collections.abc import Iterable, Mapping, Iterator |
| 7 | +import urllib.request |
| 8 | +import urllib.parse |
| 9 | +from dataclasses import dataclass |
| 10 | +from packaging.version import Version, InvalidVersion |
| 11 | +from packaging.specifiers import SpecifierSet |
| 12 | +from packaging.requirements import Requirement |
| 13 | +from functools import lru_cache |
| 14 | + |
| 15 | + |
| 16 | +github_api_url = os.environ.get("GITHUB_API_URL", "https://api.github.com") |
| 17 | + |
| 18 | + |
| 19 | +@lru_cache(maxsize=None) |
| 20 | +def _get_github_releases(repo: str, page: int): |
| 21 | + print(f"Github REST request {repo} page {page}") |
| 22 | + query = urllib.parse.urlencode({"per_page": 100, "page": page}) |
| 23 | + url = f"{github_api_url}/repos/{repo}/releases?{query}" |
| 24 | + headers = {"Accept": "application/vnd.github+json"} |
| 25 | + if os.environ.get("REST_TOKEN"): |
| 26 | + headers["Authorization"] = f"token {os.environ.get('REST_TOKEN')}" |
| 27 | + req = urllib.request.Request(url, headers=headers) |
| 28 | + with urllib.request.urlopen(req) as resp: |
| 29 | + data = json.loads(resp.read().decode()) |
| 30 | + return data |
| 31 | + |
| 32 | + |
| 33 | +@lru_cache(maxsize=None) |
| 34 | +def _get_tags(repo: str) -> Iterator[Version]: |
| 35 | + for page in itertools.count(1): |
| 36 | + data = _get_github_releases(repo, page) |
| 37 | + if not data: |
| 38 | + break |
| 39 | + for release in data: |
| 40 | + try: |
| 41 | + v = Version(release["tag_name"]) |
| 42 | + except InvalidVersion: |
| 43 | + continue |
| 44 | + else: |
| 45 | + yield v |
| 46 | + |
| 47 | + |
| 48 | +def _get_compatible_tags(repo: str, specifier: SpecifierSet) -> Iterator[Version]: |
| 49 | + return (v for v in _get_tags(repo) if v in specifier) |
| 50 | + |
| 51 | + |
| 52 | +@dataclass(frozen=True) |
| 53 | +class Library: |
| 54 | + lib_name: str |
| 55 | + repo_name: str |
| 56 | + |
| 57 | + |
| 58 | +@lru_cache(maxsize=None) |
| 59 | +def _exec_module(module_str: str) -> dict[str, Any]: |
| 60 | + m = {} |
| 61 | + exec(module_str, m, m) |
| 62 | + return m |
| 63 | + |
| 64 | + |
| 65 | +@lru_cache(maxsize=None) |
| 66 | +def _get_requirements_module(repo: str, tag: str) -> dict[str, Any]: |
| 67 | + url = f"https://raw.githubusercontent.com/{repo}/{tag}/requirements.py" |
| 68 | + with urllib.request.urlopen(url) as resp: |
| 69 | + module_str = resp.read().decode() |
| 70 | + return _exec_module(module_str) |
| 71 | + |
| 72 | + |
| 73 | +def parse_requirements(requirements: Iterable[str]) -> Mapping[str, SpecifierSet]: |
| 74 | + split_requirements = {} |
| 75 | + for req_s in requirements: |
| 76 | + req = Requirement(req_s) |
| 77 | + split_requirements[_fix_library_name(req.name)] = req.specifier |
| 78 | + return MappingProxyType(split_requirements) |
| 79 | + |
| 80 | + |
| 81 | +@lru_cache(maxsize=None) |
| 82 | +def _get_requirements(repo: str, tag: str) -> Mapping[str, SpecifierSet]: |
| 83 | + m = _get_requirements_module(repo, tag) |
| 84 | + return parse_requirements(m["get_runtime_dependencies"]()) |
| 85 | + |
| 86 | + |
| 87 | +@lru_cache(maxsize=None) |
| 88 | +def _get_pypi_releases(lib_name: str) -> MappingProxyType[str, Any]: |
| 89 | + print(f"Getting PyPI releases for {lib_name}") |
| 90 | + url = f"https://pypi.org/pypi/{lib_name}/json" |
| 91 | + with urllib.request.urlopen(url) as resp: |
| 92 | + data = json.loads(resp.read().decode()) |
| 93 | + return data["releases"] |
| 94 | + |
| 95 | + |
| 96 | +class NoValidVersion(Exception): |
| 97 | + pass |
| 98 | + |
| 99 | + |
| 100 | +@lru_cache(maxsize=None) |
| 101 | +def _get_pypi_release(lib_name: str, specifier: SpecifierSet) -> Version: |
| 102 | + releases = _get_pypi_releases(lib_name) |
| 103 | + for version_str, files in releases.items(): |
| 104 | + version = Version(version_str) |
| 105 | + # release must match the specifier and have a source distribution |
| 106 | + if version in specifier and any( |
| 107 | + file.get("packagetype", None) == "sdist" for file in files |
| 108 | + ): |
| 109 | + return version |
| 110 | + raise NoValidVersion |
| 111 | + |
| 112 | + |
| 113 | +def _fix_library_name(name: str) -> str: |
| 114 | + return name.replace("_", "-") |
| 115 | + |
| 116 | + |
| 117 | +def _find_compatible_libraries( |
| 118 | + libraries: tuple[Library, ...], |
| 119 | + requirements: Mapping[str, SpecifierSet], |
| 120 | + libraries_todo: tuple[Library, ...], |
| 121 | + libraries_frozen: Mapping[str, Version] = MappingProxyType({}), |
| 122 | +) -> Mapping[str, Version]: |
| 123 | + # Get the library to freeze and the libraries left to freeze |
| 124 | + library_freeze = libraries_todo[0] |
| 125 | + libraries_todo = tuple(libraries_todo[1:]) |
| 126 | + |
| 127 | + # Verify that this library has not already been frozen |
| 128 | + if library_freeze.lib_name in libraries_frozen: |
| 129 | + raise RuntimeError(f"Library {library_freeze.lib_name} listed more than once.") |
| 130 | + |
| 131 | + processed_requirement_configurations = [] |
| 132 | + |
| 133 | + # Iterate through all versions that match the specifier |
| 134 | + for v in _get_compatible_tags( |
| 135 | + library_freeze.repo_name, |
| 136 | + requirements.get(library_freeze.lib_name, SpecifierSet()), |
| 137 | + ): |
| 138 | + print(f"Trying {library_freeze.lib_name}=={v}") |
| 139 | + # Get the requirements this library adds |
| 140 | + library_requirements = _get_requirements(library_freeze.repo_name, str(v)) |
| 141 | + |
| 142 | + # If the library_requirements match a previous version, skip |
| 143 | + if library_requirements in processed_requirement_configurations: |
| 144 | + continue |
| 145 | + else: |
| 146 | + processed_requirement_configurations.append(library_requirements) |
| 147 | + |
| 148 | + # Extend the existing requirements. |
| 149 | + new_requirements = dict(requirements) |
| 150 | + for name, specifier in library_requirements.items(): |
| 151 | + if name in new_requirements: |
| 152 | + specifier = new_requirements[name] & specifier |
| 153 | + |
| 154 | + # check the frozen requirements are still valid. |
| 155 | + if name in libraries_frozen and libraries_frozen[name] not in specifier: |
| 156 | + raise NoValidVersion |
| 157 | + |
| 158 | + new_requirements[name] = specifier |
| 159 | + |
| 160 | + # Add the library to the frozen libraries |
| 161 | + new_libraries_frozen = {**libraries_frozen, library_freeze.lib_name: v} |
| 162 | + |
| 163 | + if libraries_todo: |
| 164 | + # if we have more libraries to freeze go to the next one |
| 165 | + try: |
| 166 | + return _find_compatible_libraries( |
| 167 | + libraries, |
| 168 | + MappingProxyType(new_requirements), |
| 169 | + libraries_todo, |
| 170 | + MappingProxyType(new_libraries_frozen), |
| 171 | + ) |
| 172 | + except NoValidVersion: |
| 173 | + # If dependency resolution failed below this, try the next version of the library |
| 174 | + continue |
| 175 | + else: |
| 176 | + # Make sure all libraries have a compatible pypi release |
| 177 | + try: |
| 178 | + for name, specifier in new_requirements.items(): |
| 179 | + if name not in new_libraries_frozen: |
| 180 | + new_libraries_frozen[name] = _get_pypi_release(name, specifier) |
| 181 | + except NoValidVersion: |
| 182 | + continue |
| 183 | + # No more libraries to freeze. |
| 184 | + return MappingProxyType(new_libraries_frozen) |
| 185 | + # Raise if no version matched |
| 186 | + raise NoValidVersion |
| 187 | + |
| 188 | + |
| 189 | +def find_compatible_libraries( |
| 190 | + libraries: Iterable[tuple[str, str]], requirements: Iterable[str] |
| 191 | +) -> Mapping[str, Version]: |
| 192 | + libraries_ = tuple( |
| 193 | + Library(_fix_library_name(lib_name), repo_name) |
| 194 | + for lib_name, repo_name in libraries |
| 195 | + ) |
| 196 | + return _find_compatible_libraries( |
| 197 | + libraries_, |
| 198 | + parse_requirements(requirements), |
| 199 | + libraries_, |
| 200 | + ) |
| 201 | + |
| 202 | + |
| 203 | +def find_and_save_compatible_libraries( |
| 204 | + compiled_library_data: Iterable[tuple[str, str]], requirements: Iterable[str] |
| 205 | +) -> None: |
| 206 | + libraries = find_compatible_libraries(compiled_library_data, requirements) |
| 207 | + print(libraries) |
| 208 | + with open( |
| 209 | + os.path.join(os.path.dirname(__file__), "libraries.json"), "w", encoding="utf-8" |
| 210 | + ) as f: |
| 211 | + json.dump({name: str(specifier) for name, specifier in libraries.items()}, f) |
0 commit comments