|
| 1 | +"""Best-effort PyPI update check. |
| 2 | +
|
| 3 | +Surfaces "a newer stackvox is on PyPI" without polluting the script paths |
| 4 | +that make up most of stackvox's invocations. The actual fetch is opt-in |
| 5 | +(only `stackvox status` and the daemon's startup trigger it); everywhere |
| 6 | +else reads from a 24h cache. |
| 7 | +
|
| 8 | +Cache schema at `~/.cache/stackvox/update-check.json`:: |
| 9 | +
|
| 10 | + {"checked_at": "2026-04-30T13:42:00+00:00", "latest": "0.4.0"} |
| 11 | +
|
| 12 | +Disable entirely with `STACKVOX_NO_UPDATE_CHECK=1`. Auto-skipped when any |
| 13 | +common CI env var is set so build logs stay clean. |
| 14 | +""" |
| 15 | + |
| 16 | +from __future__ import annotations |
| 17 | + |
| 18 | +import json |
| 19 | +import logging |
| 20 | +import os |
| 21 | +import urllib.error |
| 22 | +import urllib.request |
| 23 | +from dataclasses import dataclass |
| 24 | +from datetime import datetime, timedelta, timezone |
| 25 | +from importlib.metadata import PackageNotFoundError |
| 26 | +from importlib.metadata import version as _pkg_version |
| 27 | +from pathlib import Path |
| 28 | + |
| 29 | +from stackvox.paths import cache_dir |
| 30 | + |
| 31 | +logger = logging.getLogger(__name__) |
| 32 | + |
| 33 | + |
| 34 | +def _current_version() -> str: |
| 35 | + """Read our own installed version. Late-bound so importing this module |
| 36 | + early in the package init chain doesn't trip a circular import.""" |
| 37 | + try: |
| 38 | + return _pkg_version("stackvox") |
| 39 | + except PackageNotFoundError: |
| 40 | + return "0.0.0+unknown" |
| 41 | + |
| 42 | + |
| 43 | +PYPI_JSON_URL = "https://pypi.org/pypi/stackvox/json" |
| 44 | +CACHE_TTL = timedelta(hours=24) |
| 45 | +FETCH_TIMEOUT_SECONDS = 2.0 |
| 46 | + |
| 47 | +# Env vars set in common CI environments. Presence of any disables the check. |
| 48 | +_CI_ENV_VARS = ("CI", "GITHUB_ACTIONS", "BUILDKITE", "CIRCLECI", "GITLAB_CI", "TRAVIS") |
| 49 | + |
| 50 | + |
| 51 | +@dataclass(frozen=True) |
| 52 | +class UpdateInfo: |
| 53 | + current: str |
| 54 | + latest: str |
| 55 | + |
| 56 | + @property |
| 57 | + def is_outdated(self) -> bool: |
| 58 | + return _is_newer(self.latest, self.current) |
| 59 | + |
| 60 | + |
| 61 | +def cache_path() -> Path: |
| 62 | + return cache_dir() / "update-check.json" |
| 63 | + |
| 64 | + |
| 65 | +def is_disabled() -> bool: |
| 66 | + """Whether the update check should be skipped at all.""" |
| 67 | + if os.environ.get("STACKVOX_NO_UPDATE_CHECK"): |
| 68 | + return True |
| 69 | + return any(os.environ.get(v) for v in _CI_ENV_VARS) |
| 70 | + |
| 71 | + |
| 72 | +def _is_newer(latest: str, current: str) -> bool: |
| 73 | + """Compare two PEP-440-ish dotted version strings. |
| 74 | +
|
| 75 | + Handles the X.Y.Z and X.Y.Z+suffix forms stackvox actually publishes. |
| 76 | + Non-integer segments (e.g. an `rc1` tail) sort below numeric ones so |
| 77 | + `0.4.0` > `0.4.0rc1`, but two distinct pre-release labels compare |
| 78 | + equal — fine for stackvox since we don't ship alphas/betas. |
| 79 | + """ |
| 80 | + |
| 81 | + def _key(v: str) -> tuple[int, ...]: |
| 82 | + head = v.split("+", 1)[0] # drop +local |
| 83 | + parts: list[int] = [] |
| 84 | + for piece in head.split("."): |
| 85 | + try: |
| 86 | + parts.append(int(piece)) |
| 87 | + except ValueError: |
| 88 | + parts.append(-1) |
| 89 | + return tuple(parts) |
| 90 | + |
| 91 | + return _key(latest) > _key(current) |
| 92 | + |
| 93 | + |
| 94 | +def fetch_latest_version(timeout: float = FETCH_TIMEOUT_SECONDS) -> str | None: |
| 95 | + """Hit PyPI's JSON API for the latest stackvox version. |
| 96 | +
|
| 97 | + Returns None on any network or parse error — this is best-effort. |
| 98 | + """ |
| 99 | + if is_disabled(): |
| 100 | + return None |
| 101 | + try: |
| 102 | + ua = f"stackvox/{_current_version()} update-check" |
| 103 | + req = urllib.request.Request(PYPI_JSON_URL, headers={"User-Agent": ua}) |
| 104 | + with urllib.request.urlopen(req, timeout=timeout) as resp: # noqa: S310 - constant URL |
| 105 | + payload = json.loads(resp.read().decode("utf-8")) |
| 106 | + version = payload.get("info", {}).get("version") |
| 107 | + if isinstance(version, str): |
| 108 | + return version |
| 109 | + return None |
| 110 | + except (urllib.error.URLError, TimeoutError, OSError, ValueError) as exc: |
| 111 | + logger.debug("update check failed: %s", exc) |
| 112 | + return None |
| 113 | + |
| 114 | + |
| 115 | +def write_cache(latest: str, *, now: datetime | None = None) -> None: |
| 116 | + """Persist the most recent successful check.""" |
| 117 | + path = cache_path() |
| 118 | + path.parent.mkdir(parents=True, exist_ok=True) |
| 119 | + payload = { |
| 120 | + "checked_at": (now or datetime.now(timezone.utc)).isoformat(), |
| 121 | + "latest": latest, |
| 122 | + } |
| 123 | + path.write_text(json.dumps(payload), encoding="utf-8") |
| 124 | + |
| 125 | + |
| 126 | +def read_cache() -> tuple[datetime, str] | None: |
| 127 | + """Return (timestamp, latest) from the cache file, or None if absent/broken.""" |
| 128 | + path = cache_path() |
| 129 | + if not path.is_file(): |
| 130 | + return None |
| 131 | + try: |
| 132 | + payload = json.loads(path.read_text(encoding="utf-8")) |
| 133 | + when = datetime.fromisoformat(payload["checked_at"]) |
| 134 | + latest = payload["latest"] |
| 135 | + if not isinstance(latest, str): |
| 136 | + return None |
| 137 | + return when, latest |
| 138 | + except (OSError, ValueError, KeyError) as exc: |
| 139 | + logger.debug("ignoring malformed update-check cache: %s", exc) |
| 140 | + return None |
| 141 | + |
| 142 | + |
| 143 | +def cached_update(*, now: datetime | None = None) -> UpdateInfo | None: |
| 144 | + """Read the cache and return UpdateInfo for an unmet upgrade, else None.""" |
| 145 | + if is_disabled(): |
| 146 | + return None |
| 147 | + entry = read_cache() |
| 148 | + if entry is None: |
| 149 | + return None |
| 150 | + _checked_at, latest = entry |
| 151 | + info = UpdateInfo(current=_current_version(), latest=latest) |
| 152 | + return info if info.is_outdated else None |
| 153 | + |
| 154 | + |
| 155 | +def check_for_update(*, now: datetime | None = None) -> UpdateInfo | None: |
| 156 | + """Fetch from PyPI if the cache is stale, then return any pending update. |
| 157 | +
|
| 158 | + Synchronous; safe to call from the foreground only when ~2s of network |
| 159 | + latency is acceptable (e.g. `stackvox status`). For the daemon startup |
| 160 | + path, call this from a background thread. |
| 161 | + """ |
| 162 | + if is_disabled(): |
| 163 | + return None |
| 164 | + now = now or datetime.now(timezone.utc) |
| 165 | + entry = read_cache() |
| 166 | + if entry is None or (now - entry[0]) > CACHE_TTL: |
| 167 | + latest = fetch_latest_version() |
| 168 | + if latest is not None: |
| 169 | + write_cache(latest, now=now) |
| 170 | + else: |
| 171 | + # On fetch failure, fall back to whatever's already cached (which |
| 172 | + # may be None, in which case we just give up silently). |
| 173 | + if entry is None: |
| 174 | + return None |
| 175 | + latest = entry[1] |
| 176 | + else: |
| 177 | + latest = entry[1] |
| 178 | + info = UpdateInfo(current=_current_version(), latest=latest) |
| 179 | + return info if info.is_outdated else None |
| 180 | + |
| 181 | + |
| 182 | +def format_notice(info: UpdateInfo) -> str: |
| 183 | + """Single-line user-facing message describing the available upgrade.""" |
| 184 | + return f"update available: {info.current} → {info.latest} (run `pipx upgrade stackvox`)" |
0 commit comments