Coverage for src/edwh/local_tasks/plugin.py: 25%
483 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-13 17:03 +0200
1"""
2Extra namespace for plugin tasks such as plugin.add
3"""
5import concurrent.futures
6import datetime as dt
7import importlib
8import json
9import os
10import re
11import sys
12import typing
13from collections import OrderedDict
14from dataclasses import dataclass
15from pathlib import Path
16from typing import Optional
18import dateutil.parser
19import keyring
20import yayarl as yarl
21from ewok import (
22 Context,
23 task,
24)
25from packaging.version import parse as parse_package_version
26from termcolor import colored, cprint
27from termcolor._types import Color
29from .. import confirm, interactive_selected_radio_value, kwargs_to_options
30from ..meta import (
31 Version,
32 _gather_package_metadata_threaded,
33 _get_available_plugins_from_pypi,
34 _get_latest_version_from_pypi,
35 _parse_versions,
36 _pip,
37 is_installed,
38 pip_install,
39 pip_uninstall,
40)
41from ..release_backend import (
42 PYPROJECT,
43 Backend,
44 copy_pypi_token,
45 detect_backend,
46 pin_backend,
47 pinned_backend,
48 store_vommit_pypi_token,
49 vommit_configured,
50 vommit_specifier,
51 vommit_tasks,
52 vommit_token_complaint,
53)
56def list_installed_plugins(c: Context, pip_command: Optional[str] = None) -> list[str]:
57 """
58 List installed edwh-plugins
59 """
60 if not pip_command:
61 pip_command = _pip()
63 if result := c.run(f"{pip_command} freeze | grep -E 'edwh|ewok'", hide=True, warn=True):
64 packages = result.stdout.strip().split("\n")
65 else:
66 packages = []
68 # filter out comments and editable (local) installs:
69 regular_installs = [_ for _ in packages if not (_.startswith("#") or _.startswith("-e"))]
70 local_installs = [_.split("/")[-1] for _ in packages if _.startswith("-e")]
72 return regular_installs + local_installs
75@dataclass
76class Plugin:
77 raw_name: str
78 installed_version: typing.Optional[Version]
79 latest_version: typing.Optional[Version]
80 metadata: dict[str, typing.Any]
82 is_installed: bool
83 clean_name: str = ""
84 is_outdated: bool = False
86 def __post_init__(self) -> None:
87 if self.latest_version and self.installed_version:
88 self.is_outdated = self.latest_version > self.installed_version
90 self.clean_name = self.raw_name.removeprefix("edwh-").removesuffix("-plugin")
91 self.github_url = self.metadata["info"]["project_urls"]["Documentation"]
92 self.requires_python = self.metadata["info"]["requires_python"]
94 def __repr__(self) -> str:
95 version = (self.installed_version if self.is_installed else self.latest_version) or "?"
96 return f"<EW Plugin: {self.clean_name}-{version} {'installed' if self.is_installed else 'available'}>"
98 def __str__(self) -> str:
99 return json.dumps(self.__dict__)
101 def print_details(self, verbose: bool = False) -> None:
102 if self.is_outdated:
103 if verbose:
104 plugin_details = (
105 f"• {self.clean_name} "
106 f"({self.installed_version} < {self.latest_version}) "
107 f"- {self.github_url} "
108 f"- Python {self.requires_python}"
109 )
110 else:
111 plugin_details = (
112 f"• {self.clean_name} ({self.installed_version} < {self.latest_version}) - {self.github_url}"
113 )
115 cprint(
116 plugin_details,
117 "yellow",
118 )
119 elif self.is_installed and not self.installed_version:
120 if verbose:
121 plugin_details = f"• {self.clean_name} (unknown) - {self.github_url} - Python {self.requires_python}"
122 else:
123 plugin_details = f"• {self.clean_name} - {self.github_url}"
125 cprint(
126 plugin_details,
127 "yellow",
128 )
129 elif self.is_installed:
130 if verbose:
131 plugin_details = (
132 f"• {self.clean_name} ({self.latest_version}) - {self.github_url} - Python {self.requires_python}"
133 )
134 else:
135 plugin_details = f"• {self.clean_name} - {self.github_url}"
137 cprint(
138 plugin_details,
139 "green",
140 )
141 else:
142 if verbose:
143 plugin_details = (
144 f"◦ {self.clean_name} ({self.latest_version}) - {self.github_url} - Python {self.requires_python}"
145 )
146 else:
147 plugin_details = f"◦ {self.clean_name} - {self.github_url}"
149 cprint(
150 plugin_details,
151 "red",
152 )
155def _gather_plugin_info(c: Context, plugin_names: list[str]) -> list[Plugin]:
156 """
157 For all queried plugins (in `plugin_names`), get a Plugin instance with info.
158 """
159 installed_plugins_raw = list_installed_plugins(c)
160 installed_plugins = _parse_versions(installed_plugins_raw)
161 plugin_names = [_require_affixes(_) for _ in plugin_names]
162 plugin_infos = _gather_package_metadata_threaded(plugin_names)
164 result = []
166 for plugin_name in plugin_names:
167 metadata = plugin_infos.get(plugin_name, {})
168 if not (metadata and (info := metadata.get("info"))):
169 # invalid plugin
170 continue
172 result.append(
173 Plugin(
174 raw_name=plugin_name,
175 is_installed=plugin_name in installed_plugins,
176 installed_version=installed_plugins.get(plugin_name),
177 latest_version=parse_package_version(info["version"]),
178 metadata=metadata,
179 )
180 )
182 return result
185def gather_plugin_info(c: Context) -> list[Plugin]:
186 """
187 For all available plugins, get a Plugin instance with info
188 """
189 available_plugins = ["edwh", *_get_available_plugins_from_pypi("edwh", "plugins")]
191 installed_plugins_raw = list_installed_plugins(c)
192 if not installed_plugins_raw or (len(installed_plugins_raw) == 1 and installed_plugins_raw[0] == ""):
193 cprint("No 'edwh' packages found. That can't be right", color="yellow")
195 return _gather_plugin_info(c, available_plugins)
198@task(name="list")
199def list_plugins(c: Context, verbose: bool = False) -> None:
200 """
201 List installed plugins
203 :param c: invoke ctx
204 :type c: Context
206 :param verbose: should all info such as installed version always be shown?
207 """
208 plugins = gather_plugin_info(c)
210 old_plugins = []
211 not_all_installed: Optional[str] = None
212 for plugin in plugins:
213 plugin.print_details(verbose=verbose)
214 if plugin.is_outdated:
215 old_plugins.append(plugin)
216 if not plugin.is_installed:
217 not_all_installed = plugin.clean_name
219 if old_plugins:
220 print()
221 s = "" if len(old_plugins) == 1 else "s"
222 verb = "is" if len(old_plugins) == 1 else "are"
223 cprint(
224 f"{len(old_plugins)} plugin{s} {verb} out of date. "
225 f"Try `edwh self-update` to fix this "
226 f"or `edwh plugins --changelog` to see what's new.",
227 "yellow",
228 )
230 if not_all_installed:
231 print()
232 cprint(
233 f"Tip: not all plugins are installed. "
234 f"For example, try `edwh plugin.add {not_all_installed}` or `edwh plugin.add all`",
235 "blue",
236 )
239def _require_affixes(package: str, prefix: str = "edwh-", suffix: str = "-plugin") -> str:
240 """
241 affix is 'an addition to the base form or stem of a word in order to modify its meaning or create a new word.'
242 """
243 if package == "edwh":
244 # don't require affixes!
245 return package
247 package = package.removeprefix(prefix).removesuffix(suffix)
248 return f"{prefix}{package}{suffix}"
251@task()
252def add_all(c: Context) -> None:
253 """
254 Install all available plugins
256 Args:
257 c (Context): invoke ctx
258 """
259 plugins = _get_available_plugins_from_pypi("edwh", "plugins")
261 pip_install(c, *plugins)
264@task()
265def remove_all(c: Context) -> None:
266 """
267 Remove all available plugins
269 Args:
270 c (Context): invoke ctx
271 """
272 plugins = _get_available_plugins_from_pypi("edwh", "plugins")
274 pip_uninstall(c, *plugins)
277@task(aliases=("install",))
278def add(c: Context, plugin_names: str) -> None:
279 """
280 Install a new plugin.
282 Args:
283 c (Context): invoke ctx
284 plugin_names: which plugin(s) to add. You can add multiple plugins by separating them with a comma
285 (e.g. `edwh plugin.add restic,multipass,bundler`).
286 You can install all plugins by using 'all': `edwh plugin.add all`.
287 """
288 if plugin_names == "all":
289 return add_all(c)
291 plugin_names_splitted = [_require_affixes(plugin_name.strip()) for plugin_name in plugin_names.split(",")]
293 pip_install(c, *plugin_names_splitted)
296@task(aliases=("upgrade",))
297def update(
298 c: Context, plugin_names: str, version: Optional[str] = None, verbose: bool = False, force: bool = False
299) -> None:
300 """
301 Update a plugin (or 'all') to the latest version
303 Args:
304 c (Context): invoke ctx
305 plugin_names: the edwh plugin name (can be supplied without edwh- prefix or -plugin suffix)
306 version: optional custom version string (e.g. 0.14.0b1 for a beta pre-release)
307 verbose: show which will would be installed for each plugin
308 """
309 if force:
310 # first clean cache to ensure latest version:
311 c.run("uv cache clean", hide=True)
313 if plugin_names == "all":
314 from ..tasks import self_update
316 return self_update(c, no_cache=force)
318 plugins_with_version = []
319 for plugin_name in plugin_names.split(","):
320 plugin_name = _require_affixes(plugin_name.strip())
321 plugin_version = version or _get_latest_version_from_pypi(plugin_name)
322 plugins_with_version.append(f"{plugin_name}=={plugin_version}")
324 if verbose:
325 cprint(str(plugins_with_version), "blue")
327 pip_install(c, *plugins_with_version)
330@task(aliases=("uninstall",))
331def remove(c: Context, plugin_names: str) -> None:
332 """
333 Remove a plugin (or 'all')
335 Args:
336 c (Context): invoke ctx
337 plugin_names: which plugin to remove
338 """
339 if plugin_names == "all":
340 return remove_all(c)
342 # ensure the prefix and suffix exist, but not twice:
343 plugin_names_splitted = [_require_affixes(plugin_name.strip()) for plugin_name in plugin_names.split(",")]
345 pip_uninstall(c, *plugin_names_splitted)
348GITHUB_RAW_URL = yarl.URL("https://raw.githubusercontent.com")
351def get_changelog(github_repo: str | yarl.URL) -> str:
352 if isinstance(github_repo, str):
353 github_repo = yarl.URL(github_repo)
355 github_repo = github_repo.path.removeprefix("/") # e.g. educationwarehouse/edwh
356 changelog_url = GITHUB_RAW_URL / github_repo / "master/CHANGELOG.md" # replace github.com with github raw
358 return changelog_url.get(timeout=10).text
361def get_changelogs_threaded(github_repos: dict[str, str]) -> dict[str, str]:
362 """
363 For any package in packages, gather its metadata from pypi
364 """
365 all_data: dict[str, str] = {}
366 with concurrent.futures.ThreadPoolExecutor() as executor:
367 repo_urls = list(github_repos.values())
368 for result, package in zip(executor.map(get_changelog, repo_urls), github_repos.keys()):
369 all_data[package] = result
371 return all_data
374def _filter_away_version(changelog_version: Version, _filter: str) -> bool:
375 """
376 If _filter is a Version and it's bigger than the selected row (via 'changelog_version'),
377 the row should not be visible.
378 """
379 try:
380 filter_version = parse_package_version(_filter)
381 return changelog_version <= filter_version
382 except Exception:
383 return False
386def _filter_away_date(date: dt.datetime, _filter: str) -> bool:
387 """
388 If _filter is a date and it's bigger than the selected row (via 'date'), the row should not be visible.
389 """
390 try:
391 return date <= dateutil.parser.parse(_filter)
392 except Exception:
393 return False
396def _filter_away(version: Version, date: dt.datetime, _filter: str) -> bool:
397 """
398 If a filter is numeric, it's probably not a version or date (it could be parsed as one but we want other behavior).
399 If it is not numeric, try filtering away low version or dates.
401 Returns True if a row can be removed and False if it has to stay.
402 """
403 return (not _filter.isnumeric()) and (_filter_away_version(version, _filter) or _filter_away_date(date, _filter))
406def sort_versions(key_value: tuple[str, typing.Any]) -> Version:
407 """
408 Can be used as key=sort_versions in sort_and_filter_changelog
409 """
410 key, _ = key_value
412 try:
413 version, _date = key.split(" ")
414 return parse_package_version(version)
415 except Exception:
416 # something went wrong, return something so sorting doesn't crash:
417 return Version("0.0.0")
420type T_Changelog = dict[str, dict[str, list[str]]]
421type T_OrderedChangelog = OrderedDict[str, dict[str, list[str]]]
422type T_OrderedChangelogs = dict[str, T_OrderedChangelog]
423type T_Changelogs = dict[str, T_Changelog]
426def parse_changelog(markdown: str) -> T_Changelog:
427 """
428 Parse our CHANGELOG.md to a dictionary of {version: {type: [list of changes]}}
429 where version is e.g. v0.18.5 (2023-06-06)
430 where type is e.g. Fix
431 """
432 # thanks ChatGPT
433 changelog: dict[str, dict[str, list[str]]] = {}
434 current_version: Optional[str] = None
435 current_category: Optional[str] = None
437 lines = markdown.split("\n")
438 for line in lines:
439 if line.startswith("# Changelog"):
440 continue
442 version_match = re.match(r"^## (.+)", line)
443 if version_match:
444 version = version_match.group(1)
445 changelog[version] = {}
446 current_version = version
447 continue
449 category_match = re.match(r"^### (.+)", line)
450 if category_match and current_version:
451 category = category_match.group(1)
452 changelog[current_version][category] = []
453 current_category = category
454 continue
456 feature_match = re.match(r"^\* (.+)", line)
457 if feature_match and current_version and current_category:
458 feature = feature_match.group(1)
459 changelog[current_version][current_category].append(feature)
461 return changelog
464def to_date(key: str) -> dt.datetime:
465 """
466 Convert a changelog key `v0.0.0 (2000-01-01)` to a dt.datetime
467 """
468 try:
469 _, date = key.split(" ", 1)
470 return dateutil.parser.parse(date.removeprefix("(").removesuffix(")"))
471 except Exception:
472 return dateutil.parser.parse("2000-01-01")
475def to_version(key: str) -> Version:
476 """
477 Convert a changelog key `v0.0.0 (2000-01-01)` to a Version(0.0.0)
478 """
479 try:
480 key, _ = key.split(" ", 1)
481 return parse_package_version(key)
482 except Exception:
483 return Version("0.0.0")
486def sort_and_filter_changelog(changelog: dict[str, dict[str, list[str]]], since: Optional[str] = None) -> T_Changelog:
487 """
488 Since can be:
489 - a number - amount of releases to show.
490 - a version number - show changes starting from that version.
491 - a date - show changes starting from that date.
492 - major, minor, patch - show changes starting from the latest release of that type.
493 """
494 filtered = {}
496 prev_major = prev_minor = prev_patch = 0
498 for idx, (k, v) in enumerate(changelog.items()):
499 version = to_version(k)
500 date = to_date(k)
502 # checks to stop:
503 if since and (
504 (since == "major" and version.major < prev_major)
505 or (since == "minor" and (version.minor < prev_minor or version.major < prev_major))
506 or (
507 since == "patch"
508 and (version.micro < prev_patch or version.minor < prev_minor or version.major < prev_major)
509 )
510 or (since.isnumeric() and idx >= int(since))
511 ):
512 break
514 # checks to skip:
515 elif since and _filter_away(version, date, since):
516 # skip!
517 continue
519 prev_major = version.major
520 prev_minor = version.minor
521 prev_patch = version.micro
522 # checks passed, add to output
523 filtered[k] = v
525 return OrderedDict(sorted(filtered.items(), reverse=True, key=sort_versions))
528COLORS: dict[str, Color] = {
529 "fix": "yellow",
530 "feature": "green",
531 "documentation": "blue",
532}
534BOLD_RE = re.compile(r"((\*\*|__).+?(\*\*|__))")
537def colored_markdown(text: str) -> str:
538 """
539 Prettify a changelog line (makes ** bold).
541 todo: more than bold?
542 """
543 final = ""
544 for part in BOLD_RE.split(text):
545 if part.startswith("**") and part.endswith("**"):
546 part = colored(part.removeprefix("**").removesuffix("**"), attrs=["bold"])
547 final += part
548 return final
551def display_changelogs(changelogs: T_Changelogs) -> None:
552 """
553 Final step of changelog(), uses the result of {package: sort_and_filter_changelog()}.
554 """
555 for package, history in changelogs.items():
556 cprint(package, "red", attrs=["bold", "underline"])
557 for version, changes in history.items():
558 print("-", version)
559 for change_type, change_descriptions in changes.items():
560 print("--", colored(change_type, COLORS.get(change_type.lower(), "white")))
561 for change in change_descriptions:
562 print("----", colored_markdown(change))
565def _gather_and_display_changelogs(info: list[Plugin], since: dict[str, str]) -> None:
566 changelogs_raw = get_changelogs_threaded(
567 {plugin.clean_name: plugin.metadata["info"]["project_urls"]["Source"] for plugin in info}
568 )
570 changelogs_parsed: T_Changelogs = {
571 name: (
572 # sort and filter removes everything not matching 'since' and sorts by date (/version) desc.
573 sort_and_filter_changelog(
574 # parse_changelog converts the markdown to a dict
575 parse_changelog(data),
576 # 'since' filter can differ per plugin if --new is passed.
577 since[name],
578 )
579 )
580 for name, data in changelogs_raw.items()
581 }
583 display_changelogs(changelogs_parsed)
586def _changelog_new(ctx: Context, *_: typing.Any) -> None:
587 """
588 List changes since last installed version.
589 """
590 info = [plugin for plugin in gather_plugin_info(ctx) if plugin.is_outdated]
591 # if --new, ignore --since argument
592 since = {plugin.clean_name: str(plugin.installed_version) for plugin in info}
594 return _gather_and_display_changelogs(info, since)
597def _changelog_specific(ctx: Context, plugin_names: list[str], since: str, *_: typing.Any) -> None:
598 """
599 List changes for specific plugins.
600 """
601 info = _gather_plugin_info(ctx, plugin_names)
602 _since = {plugin.clean_name: since for plugin in info}
604 return _gather_and_display_changelogs(info, _since)
607def _changelog_all(ctx: Context, _: list[str], since: str, *__: typing.Any) -> None:
608 """
609 List changes for all plugins.
610 """
611 info = gather_plugin_info(ctx)
612 _since = {plugin.clean_name: since for plugin in info}
614 return _gather_and_display_changelogs(info, _since)
617@task(iterable=["plugin"])
618def changelog(ctx: Context, plugin: list[str], since: str = "5", new: bool = False) -> None:
619 """
620 Show changelogs for edwh plugins.
621 by default, changelogs from all plugins are shown.
622 Since can be used to filter/limit changes. By default, the last 5 releases are shown.
623 Since can be a number (amount of changes), a date (show releases from that date),
624 a version (releases starting from that version) or
625 'major'/'minor'/'patch' to show releases since the latest version of that type.
626 if 'new' is True, show only changes for outdated packages.
627 """
628 if new:
629 return _changelog_new(ctx, plugin, since, new)
630 elif plugin:
631 return _changelog_specific(ctx, plugin, since, new)
632 else:
633 return _changelog_all(ctx, plugin, since, new)
636def _semantic_release_publish(c: Context, flags: dict[str, typing.Any], **kw: typing.Any) -> typing.Optional[str]:
637 """
638 Run the deprecated python-semantic-release path.
640 Kept for projects that have not moved to vommit yet; see `release`.
641 """
642 semver = c.run(f"semantic-release publish {kwargs_to_options(flags)}", **kw)
644 matches: list[str] = re.findall(r"to (\d+\.\d+\.\d+.*)", semver.stderr if semver else "")
645 if new_version := matches:
646 return new_version[0]
648 cprint("No new version found!", "yellow")
649 return None
652def uvenv(ctx: Context, specifier: str):
653 """
654 Install something using uvenv.
656 specifier can be a package name, optionally with version specifier:
657 `uvenv(ctx, 'python-semantic-release<8')`
658 """
659 return ctx.run(f"~/.local/bin/uvenv install '{specifier}'", warn=True)
662@task()
663def require_semantic_release(ctx: Context):
664 """
665 Task to ensure psr is available.
667 Part of the deprecated release path; new projects use vommit instead.
668 """
669 if is_installed(ctx, "semantic-release"):
670 return
672 uvenv(ctx, "python-semantic-release<8")
674 assert is_installed(ctx, "semantic-release"), "Tool 'semantic-release' still can't be found!"
677PSR_DEPRECATION = (
678 "python-semantic-release support is deprecated and will be removed in edwh 2.0. "
679 "Run `edwh plugin.release` again to migrate this project to vommit."
680)
683def _vommit_spec() -> str:
684 """
685 What to install, taking edwh's keyring backend into account.
687 A plain `vommit` cannot read or write the ssh-agent-backed keyring, so a
688 project relying on it would install vommit and still be unable to reach its
689 own PyPI token.
690 """
691 # local: ..tasks imports this module's package, so a top-level import cycles
692 from ..tasks import ssh_agent_keyring_config_path
694 extras = "[ssh]" if ssh_agent_keyring_config_path().exists() else ""
696 return f"vommit{extras}{vommit_specifier()}"
699def ensure_vommit(ctx: Context) -> bool:
700 """
701 Ensure vommit is importable, offering to install it when it isn't.
703 Installs into edwh's own environment rather than via uvenv: that is what
704 activates vommit's `edwh` entry point, so `edwh vommit.*` starts working
705 too.
707 A plain function rather than a task, because ewok writes a task's return
708 value into the shared `ctx["result"]`, and the enclosing task then returns
709 that instead of its own -- which would make `plugin.bump` answer False.
710 """
711 if vommit_tasks(): 711 ↛ 712line 711 didn't jump to line 712 because the condition on line 711 was never true
712 return True
714 spec = _vommit_spec()
715 if not confirm(f"vommit is not installed. Install {spec} now? [Yn] ", default=True):
716 return False
718 pip_install(ctx, spec)
720 # site-packages is already on sys.path, so the import finder just needs to
721 # be told to look again.
722 importlib.invalidate_caches()
723 if vommit_tasks(): 723 ↛ 724line 723 didn't jump to line 724 because the condition on line 723 was never true
724 return True
726 cprint("vommit was installed but is not importable yet; please run this command again.", "yellow")
727 return False
730@task()
731def require_vommit(ctx: Context) -> None:
732 """
733 Install vommit, the release backend, if this environment lacks it.
734 """
735 ensure_vommit(ctx)
738SWITCH_NOW = "now"
739SWITCH_LATER = "later"
740SWITCH_NEVER = "never"
742MIGRATE_OPTIONS = {
743 SWITCH_NOW: "migrate now - walk through vommit's migrator, keeping your v7 settings",
744 SWITCH_LATER: "not now - release with python-semantic-release this time, ask again next time",
745 SWITCH_NEVER: "never - keep this project on python-semantic-release and stop asking",
746}
748SETUP_OPTIONS = {
749 SWITCH_NOW: "set up now - configure vommit for this project",
750 SWITCH_LATER: "not now - ask again next time",
751 SWITCH_NEVER: "never - stop asking about this project",
752}
755def _can_ask() -> bool:
756 """
757 Whether there is anybody to answer a radio prompt.
759 `confirm` honours EDWH_NON_INTERACTIVE itself, but the radio helper reads
760 the terminal directly and would hang or misread without this guard.
761 """
762 return os.environ.get("EDWH_NON_INTERACTIVE", "0") != "1" and sys.stdin.isatty()
765def _offer_switch(c: Context, backend: Backend, pyproject: Path) -> Backend:
766 """
767 Offer to move this project to vommit, and report which backend to use now.
769 Returns "vommit" only when a config was actually written: vommit's migrator
770 can be stopped halfway on purpose, and this release has to fall back rather
771 than hand over to a config that never landed.
772 """
773 migrating = backend == "psr"
775 if migrating: 775 ↛ 776line 775 didn't jump to line 776 because the condition on line 775 was never true
776 cprint("This project still releases with python-semantic-release.", "blue")
777 else:
778 cprint("This project has no release configuration yet.", "blue")
780 if not _can_ask(): 780 ↛ 785line 780 didn't jump to line 785 because the condition on line 780 was always true
781 if not migrating: 781 ↛ 783line 781 didn't jump to line 783 because the condition on line 781 was always true
782 cprint("Run `edwh plugin.release` interactively to set up vommit.", "blue")
783 return backend
785 prompt = "Switch this project to vommit?" if migrating else "Set this project up with vommit?"
786 answer = interactive_selected_radio_value(
787 MIGRATE_OPTIONS if migrating else SETUP_OPTIONS,
788 prompt=prompt,
789 selected=SWITCH_NOW,
790 )
792 if answer == SWITCH_NEVER:
793 pin_backend("psr" if migrating else "vommit", pyproject)
794 cprint(f"Recorded your choice in {pyproject}; edwh will not ask again.", "blue")
795 return backend
796 elif answer != SWITCH_NOW:
797 # "not now", or the prompt was abandoned
798 return backend
799 elif not ensure_vommit(c):
800 return backend
802 if migrating:
803 return _migrate_to_vommit(c, pyproject)
805 return _setup_vommit(c, pyproject)
808def _migrate_to_vommit(c: Context, pyproject: Path) -> Backend:
809 """
810 Copy the PyPI token across, then hand over to vommit's migrator.
811 """
812 copy_pypi_token()
814 tasks = vommit_tasks()
815 assert tasks, "ensure_vommit returned True without vommit being importable"
816 tasks.migrate(c, project_dir=str(pyproject.parent))
818 if not vommit_configured(pyproject):
819 cprint("Migration did not complete; releasing with python-semantic-release for now.", "yellow")
820 return "psr"
822 return "vommit"
825def _setup_vommit(c: Context, pyproject: Path) -> Backend:
826 """
827 Run vommit's interactive setup on a project with no release config.
828 """
829 tasks = vommit_tasks()
830 assert tasks, "ensure_vommit returned True without vommit being importable"
831 tasks.setup(c, project_dir=str(pyproject.parent))
833 if not vommit_configured(pyproject):
834 cprint("vommit was not configured; nothing to release with.", "yellow")
835 return "none"
837 return "vommit"
840def _resolve_backend(c: Context, pyproject: Path = PYPROJECT) -> Optional[Backend]:
841 """
842 Which backend releases this project, asking about a switch when relevant.
844 The single funnel for `release` and `bump`, so the deprecation notice lands
845 here once rather than at every place that could reach the psr path. None
846 means this project cannot be released and the reason has been reported.
847 """
848 backend = detect_backend(pyproject)
850 if backend == "vommit":
851 # vommit is an optional extra, so its config outlives its install: a
852 # migrated project on a second machine has the one without the other.
853 if vommit_tasks():
854 return backend
856 cprint(f"{pyproject} is configured for vommit, but vommit is not installed.", "yellow")
857 return backend if ensure_vommit(c) else None
859 if pinned_backend(pyproject) is None: 859 ↛ 862line 859 didn't jump to line 862 because the condition on line 859 was always true
860 backend = _offer_switch(c, backend, pyproject)
862 if backend == "psr": 862 ↛ 863line 862 didn't jump to line 863 because the condition on line 862 was never true
863 cprint(PSR_DEPRECATION, "yellow")
865 return backend
868@task()
869def require_hatch(ctx: Context):
870 """
871 Task to ensure hatch is available.
873 Part of the deprecated release path; vommit projects set
874 [tool.vommit.commands] build/publish instead of passing --hatch.
875 """
876 if is_installed(ctx, "hatch"):
877 return
879 uvenv(ctx, "hatch")
881 assert is_installed(ctx, "hatch"), "Tool 'hatch' still can't be found!"
884@dataclass
885class GitError(Exception):
886 reason: str
889@task()
890def git_pull(c: Context, yes: bool) -> None:
891 cprint("pulling latest version from git", "blue")
893 # Check for unstaged changes
894 git_status = c.run("git status --porcelain", hide=True)
895 # --porcelain produces an easier output format which empty if there are no uncommitted changes.
896 if git_status.stdout.strip():
897 cprint("Warning: You have unstaged changes in your working directory:", "yellow")
898 c.run("git status", hide=False) # Show status to help user see unstaged changes
899 if not yes and not confirm("Continue with git pull despite unstaged changes? [yN] ", default=False):
900 cprint("Operation cancelled. Please commit or stash your changes first.", "red")
901 raise GitError("unstaged changes")
903 # 1. pull
904 git_pull = c.run("git pull", warn=True)
906 # 2. check if merge is going on, in that case: stop and let the user fix it
907 if git_pull.stderr and ("merge" in git_pull.stderr.lower() or "conflict" in git_pull.stderr.lower()):
908 cprint("Git merge conflict detected! Please resolve the conflicts manually and try again.", "red")
909 c.run("git status", hide=False) # Show status to help user identify conflicting files
910 raise GitError("merge required")
912 # 3. if no merge - we good so continue
913 if git_pull.ok:
914 cprint("Git pull completed successfully", "green")
915 else:
916 cprint(f"Git pull failed: {git_pull.stderr}", "red")
917 if not yes and not confirm("Continue despite git pull failure? [yN] ", default=False):
918 raise GitError(git_pull.stderr)
921def build(c: Context, hatch: bool = False) -> list[str]:
922 if hatch:
923 hatch_build = c.run("hatch build -c")
924 else:
925 c.run("rm -r dist/ || true", hide=True)
926 hatch_build = c.run("uv build")
928 # not compiled since this isn't used a lot
929 return re.findall(r"dist/(.+)-\d+\.\d+\.\d+.+tar\.gz", hatch_build.stderr if hatch_build else "")
932@task()
933def authenticate(_: Context):
934 """
935 Store a PyPI token for releasing.
937 Written to both edwh's and vommit's keyring entries, so the token works
938 whichever backend a project uses.
939 """
940 from ..tasks import ensure_keyring_unlocked
942 pypi_token = input("Enter your token (starting with pypi-): ").strip()
943 if not pypi_token:
944 cprint("No token specified, exiting", "red")
945 exit(1)
947 if complaint := vommit_token_complaint(pypi_token):
948 cprint(complaint, "yellow")
950 ensure_keyring_unlocked()
951 keyring.set_password("edwh", "pypi", pypi_token)
953 if store_vommit_pypi_token(pypi_token):
954 cprint("Stored the token for both edwh and vommit.", "green")
955 else:
956 cprint("Stored the token for edwh; run `vommit authenticate` too once vommit is installed.", "blue")
958 return pypi_token
961def publish(c: Context, hatch: bool = False):
962 """
963 Upload a build, as the deprecated psr path does it.
965 Only reachable from `release`'s psr branch, which prints PSR_DEPRECATION
966 before it gets here; vommit projects publish via [tool.vommit.commands].
967 """
968 if hatch:
969 c.run("hatch publish")
970 else:
971 from ..tasks import ensure_keyring_unlocked
973 # without this a locked keyring raises instead of offering the ssh-agent
974 # fallback, which is exactly when a release needs the token most
975 ensure_keyring_unlocked()
977 pypi_token = keyring.get_password("edwh", "pypi")
978 if not pypi_token:
979 pypi_token = authenticate(c)
981 result = c.run("uv publish", env=dict(UV_PUBLISH_TOKEN=pypi_token), pty=True, warn=True)
983 if not result.ok and "403" in result.stdout + result.stderr:
984 # currently this message is printed to stdout but check both in case it changes (in uv)
985 cprint("Hint: you may want to enter a new token via `edwh plugin.authenticate`", "blue")
988def _psr_bump(
989 c: Context,
990 major: bool = False,
991 minor: bool = False,
992 patch: bool = False,
993 prerelease: bool = False,
994 noop: bool = False,
995 hide: bool = False,
996) -> Optional[str]:
997 """
998 Bump via python-semantic-release, installing it first if needed.
999 """
1000 # not a `pre=` on bump/release any more: a pre-task runs whatever the
1001 # project is configured for, so a vommit project installed psr to release
1002 require_semantic_release(c)
1004 return _semantic_release_publish(
1005 c,
1006 {
1007 "noop": noop,
1008 "major": major,
1009 "minor": minor,
1010 "patch": patch,
1011 "prerelease": prerelease,
1012 },
1013 hide=hide,
1014 )
1017@task()
1018def bump(
1019 c: Context,
1020 major: bool = False,
1021 minor: bool = False,
1022 patch: bool = False,
1023 prerelease: bool = False,
1024 noop: bool = False,
1025 hide: bool = False,
1026) -> Optional[str]:
1027 """
1028 Bump this project's version, using whichever release tool it is configured for.
1029 """
1030 backend = _resolve_backend(c)
1032 if backend is None:
1033 return None
1034 elif backend == "vommit": 1034 ↛ 1035line 1034 didn't jump to line 1035 because the condition on line 1034 was never true
1035 tasks = vommit_tasks()
1036 assert tasks, "backend resolved to vommit without vommit being importable"
1037 return tasks.bump(
1038 c,
1039 major=major,
1040 minor=minor,
1041 patch=patch,
1042 prerelease=prerelease,
1043 noop=noop,
1044 )
1045 elif backend == "none": 1045 ↛ 1051line 1045 didn't jump to line 1051 because the condition on line 1045 was always true
1046 # falling through would install psr and run it against a project that
1047 # has no psr config to run against
1048 cprint("No release configuration; nothing to bump.", "yellow")
1049 return None
1050 else:
1051 return _psr_bump(
1052 c,
1053 major=major,
1054 minor=minor,
1055 patch=patch,
1056 prerelease=prerelease,
1057 noop=noop,
1058 hide=hide,
1059 )
1062def _vommit_release(
1063 c: Context,
1064 hatch: bool,
1065 major: bool,
1066 minor: bool,
1067 patch: bool,
1068 prerelease: bool,
1069 noop: bool,
1070 yes: bool,
1071) -> None:
1072 """
1073 Hand a release to vommit.
1074 """
1075 if hatch:
1076 cprint(
1077 "--hatch has no meaning for a vommit project: set "
1078 "[tool.vommit.commands] build/publish instead (e.g. `hatch build -c` / `hatch publish`).",
1079 "red",
1080 )
1081 return
1083 tasks = vommit_tasks()
1084 assert tasks, "backend resolved to vommit without vommit being importable"
1085 tasks.release(
1086 c,
1087 major=major,
1088 minor=minor,
1089 patch=patch,
1090 prerelease=prerelease,
1091 noop=noop,
1092 yes=yes,
1093 )
1096@task(aliases=("publish",))
1097def release(
1098 c: Context,
1099 noop: bool = False,
1100 major: bool = False,
1101 minor: bool = False,
1102 patch: bool = False,
1103 prerelease: bool = False,
1104 yes: bool = False,
1105 pull: bool = True,
1106 hatch: bool = False,
1107) -> None:
1108 """
1109 Release a new version of a plugin.
1111 Args:
1112 c (Context)
1113 noop: don't actually publish anything, just show what would happen
1114 major: bump major version
1115 minor: bump minor version
1116 patch: bump patch version
1117 prerelease: release as beta version (e.g. 1.0.0b1)
1118 pull: it's recommended to do a git pull before trying to bump the version;
1119 otherwise the git tags could get messed up
1120 yes: don't ask for confirmation
1121 hatch: backwards-compatibility for when 'uv' doesn't work.
1122 """
1123 if pull: 1123 ↛ 1124line 1123 didn't jump to line 1124 because the condition on line 1123 was never true
1124 try:
1125 git_pull(c, yes=yes)
1126 except GitError:
1127 # stop
1128 return
1130 # git_pull runs first on both paths: vommit only fetches for its branch
1131 # check, so resolving the backend before pulling would change what --pull
1132 # means for a project that migrates during this very run.
1133 backend = _resolve_backend(c)
1135 if backend is None: 1135 ↛ 1137line 1135 didn't jump to line 1137 because the condition on line 1135 was always true
1136 return
1137 elif backend == "vommit":
1138 return _vommit_release(
1139 c,
1140 hatch=hatch,
1141 major=major,
1142 minor=minor,
1143 patch=patch,
1144 prerelease=prerelease,
1145 noop=noop,
1146 yes=yes,
1147 )
1148 elif backend == "none":
1149 cprint("No release configuration; nothing to release with.", "yellow")
1150 return
1152 if hatch:
1153 require_hatch(c)
1155 cprint("bumping version", "blue")
1157 if not (yes or noop):
1158 new_version = _psr_bump(
1159 c,
1160 major=major,
1161 minor=minor,
1162 patch=patch,
1163 prerelease=prerelease,
1164 noop=True,
1165 hide=True,
1166 )
1168 if not new_version or not confirm(
1169 f"Are you sure you would like to release version {new_version}? [yN] ", default=False
1170 ):
1171 print("bye!")
1172 return
1174 new_version = _psr_bump(
1175 c,
1176 major=major,
1177 minor=minor,
1178 patch=patch,
1179 prerelease=prerelease,
1180 noop=noop,
1181 hide=False,
1182 )
1184 if not new_version:
1185 return
1187 cprint("Starting build", "blue")
1189 pkg = build(c, hatch=hatch)
1191 if not noop:
1192 cprint("Starting release", "blue")
1193 publish(c, hatch=hatch)
1194 cprint(f"{pkg} {new_version} released!", "green")
1195 else:
1196 cprint(f"Not publishing {pkg} {new_version} due to --noop", "yellow")