Coverage for src/edwh/meta.py: 21%

1""" 

2This files contains everything to do with meta-tasks such as self-updating 

3""" 

4 

5import concurrent.futures 

6import shlex 

7import sys 

8import typing as t 

9 

10import yayarl as yarl 

11from ewok import Context, task 

12from invoke.runners import Result 

13from packaging.version import InvalidVersion, Version 

14from packaging.version import parse as parse_package_version 

15from termcolor import cprint 

16 

17from .helpers import AnyDict 

18 

19PYPI_URL_BASE = yarl.URL("https://pypi.python.org/pypi/") 

20 

21 

22def _python() -> str: 

23 """ 

24 used to detect current Python environment, even in pipx 

25 """ 

26 return sys.executable 

27 

28 

29def _pip(python: str = _python()) -> str: 

30 """ 

31 used to detect current pip environment, even in pipx 

32 """ 

33 # uv.find_uv_bin() does not really work here, because then the right venv may not be used! 

34 return f"{python} -m uv pip" 

35 

36 

37def pip_install(c: Context, *specifiers: str, **kw: t.Any) -> t.Optional[Result]: 

38 """ 

39 Install into the environment edwh itself runs in. 

40 """ 

41 return c.run(f"{_pip()} install {shlex.join(specifiers)}", **kw) 

42 

43 

44def pip_uninstall(c: Context, *specifiers: str, **kw: t.Any) -> t.Optional[Result]: 

45 """ 

46 Remove from the environment edwh itself runs in. 

47 """ 

48 return c.run(f"{_pip()} uninstall {shlex.join(specifiers)}", **kw) 

49 

50 

51def _get_pypi_info(package: str) -> AnyDict: 

52 """ 

53 Load metadata from pypi for a package 

54 """ 

55 url = PYPI_URL_BASE / package / "json" 

56 resp = url.get(timeout=10) 

57 return t.cast(AnyDict, resp.json()) 

58 

59 

60def _get_latest_version_from_pypi(package: str) -> Version: 

61 """ 

62 Get the latest Version for a package from pypi 

63 """ 

64 data = _get_pypi_info(package) 

65 if not data or not data.get("info"): 

66 raise ModuleNotFoundError(f"Plugin {package} does not seem to exist.") 

67 

68 return parse_package_version(data["info"]["version"]) 

69 

70 

71def _get_available_plugins_from_pypi(package: str, extra: str | None = None) -> list[str]: 

72 """ 

73 List all plugins available for package, optionally for a specific 'extra'. 

74 

75 e.g. 

76 [mypackage] 

77 dev = ['package1', 'package2'] 

78 another_extra = ['package1', 'package3'] 

79 

80 > _get_available_plugins_from_pypi('mypackage') 

81 ['package1', 'package2', 'package3'] 

82 

83 > _get_available_plugins_from_pypi('mypackage, 'dev') 

84 ['package1', 'package2'] 

85 

86 """ 

87 data = _get_pypi_info(package) 

88 extras = data["info"]["requires_dist"] 

89 

90 if extra: 

91 extras = [_.split(";")[0] for _ in extras if _.endswith(f'; extra == "{extra}"')] 

92 

93 return list(extras) 

94 

95 

96def _gather_package_metadata_threaded(packages: t.Iterable[str]) -> dict[str, AnyDict | None]: 

97 """ 

98 For any package in packages, gather its metadata from pypi 

99 """ 

100 all_data: dict[str, AnyDict | None] = {} 

101 with concurrent.futures.ThreadPoolExecutor() as executor: 

102 pkg_names = [_.split("==")[0] for _ in packages] 

103 for result, package in zip(executor.map(_get_pypi_info, pkg_names), packages): 

104 all_data[package] = result 

105 

106 return all_data 

107 

108 

109def _determine_newest_version(releases: t.Collection[str]) -> str: 

110 sorted_releases = sorted(releases, key=Version) 

111 return sorted_releases[-1] 

112 

113 

114def _determine_outdated_threaded(installed_plugins: t.Collection[str], prerelease: bool = False) -> dict[str, Version]: 

115 """ 

116 Like _determine_outdated but parallelized with Threading 

117 

118 installed_plugins is a list (or other iterable) of ["name==version", "name @ location"] type strings 

119 """ 

120 plugins_metadata = _gather_package_metadata_threaded([_ for _ in installed_plugins if " @ " not in _]) 

121 

122 outdated = {} 

123 for plugin, metadata in plugins_metadata.items(): 

124 if not metadata: 

125 continue 

126 

127 try: 

128 name, current_version_str = plugin.split("==") 

129 current_version = parse_package_version(current_version_str) 

130 

131 latest_stable = metadata["info"]["version"] 

132 latest_prerelease = _determine_newest_version(metadata["releases"].keys()) if prerelease else None 

133 if not (latest_stable or latest_prerelease): 

134 continue 

135 

136 latest_version = parse_package_version(t.cast(str, latest_prerelease if prerelease else latest_stable)) 

137 except Exception: 

138 # no current or latest version found? skip 

139 continue 

140 

141 if current_version and latest_version and latest_version > current_version: 

142 outdated[name] = latest_version 

143 

144 return outdated 

145 

146 

147def _parse_versions(installed: list[str]) -> dict[str, Version | None]: 

148 """ 

149 Given a list of installed packages from pip freeze (_plugins), gather the parsed versions 

150 """ 

151 versions = {} 

152 for pkg in installed: 

153 parts = pkg.split(" @ ")[0].split("==") 

154 name = parts[0] 

155 try: 

156 version = parse_package_version(parts[1]) 

157 except (InvalidVersion, IndexError): 

158 version = None 

159 # finally: 

160 versions[name] = version 

161 

162 return versions 

163 

164 

165@task() 

166def plugins(c: Context, verbose: bool = False, changelog: bool = False) -> None: 

167 """ 

168 alias for plugin.list or plugin.changelog --new 

169 """ 

170 from .local_tasks import plugin 

171 

172 if changelog: 

173 return plugin.changelog(c, [], new=True) 

174 else: 

175 return plugin.list_plugins(c, verbose=verbose) 

176 

177 

178def _self_update(c: Context, prerelease: bool = False, no_cache: bool = False) -> None: 

179 """ 

180 Wrapper for self-update that can handle type hint Context 

181 """ 

182 from .local_tasks.plugin import list_installed_plugins 

183 

184 pip_command = _pip() 

185 

186 edwh_packages = list_installed_plugins(c, pip_command) 

187 if not edwh_packages or (len(edwh_packages) == 1 and edwh_packages[0] == ""): 

188 cprint("No 'edwh' packages found. That can't be right", color="yellow") 

189 

190 old_plugins = _determine_outdated_threaded(edwh_packages, prerelease=prerelease) 

191 

192 if not old_plugins and not no_cache: 

193 cprint("Nothing to update", "blue") 

194 exit() 

195 

196 if no_cache and not old_plugins: 

197 # Fresh mode should still run for currently-up-to-date packages so transitive dependencies can be refreshed. 

198 target_packages: dict[str, Version] = { 

199 package.split("==")[0]: parse_package_version(package.split("==")[1]) 

200 for package in edwh_packages 

201 if "==" in package and " @ " not in package 

202 } 

203 else: 

204 target_packages = old_plugins 

205 

206 cprint(f"Will try to update {len(target_packages)} packages.", "blue") 

207 

208 success = [] 

209 failure = [] 

210 for plugin, version in target_packages.items(): 

211 command = f"{pip_command} install {plugin}=={version}" 

212 if no_cache: 

213 # In "fresh" mode, also refresh transitive dependencies to newest compatible versions. 

214 command = f"{command} --no-cache --upgrade --resolution highest" 

215 

216 result = c.run(command, warn=True) 

217 

218 if result and result.return_code == 0: 

219 success.append(plugin) 

220 else: 

221 failure.append(plugin) 

222 

223 if success: 

224 cprint(f"{len(success)}/{len(target_packages)} updated successfully.", "green") 

225 

226 if failure: 

227 cprint(f"{', '.join(failure)} failed updating", "red") 

228 

229 

230@task( 

231 flags={ 

232 "prerelease": ["prerelease", "pre", "pre-release", "p"], 

233 "no_cache": ["no-cache", "f", "fresh"], 

234 } 

235) 

236def self_update(c: Context, prerelease: bool = False, no_cache: bool = False) -> None: 

237 """Updates `edwh` and all installed plugins. 

238 

239 Args: 

240 c (Context): invoke ctx 

241 prerelease (bool, optional): allow non-stable releases? Defaults to False. 

242 no_cache (bool, optional): download fresh? Defaults to False. 

243 """ 

244 return _self_update(c, prerelease, no_cache) 

245 

246 

247def is_installed(ctx: Context, command: str) -> bool: 

248 """ 

249 Check if a bash command is known. 

250 """ 

251 result = ctx.run(f"which {command}", hide="both", warn=True) 

252 return bool(result and result.ok)