orisys-tech/orisys-sdk
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix pypi cannot check after upload

Browse Source
This commit is contained in:
ykcheungab
2026-09-12 17:43:40 +08:00
parent 488c8b0bc1
commit 5f78115429
3 changed files with 237 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
build/release/pypi.py
+113 -14
View File
@@ -22,6 +22,7 @@ import shutil
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.request
from dataclasses import dataclass, field
@@ -182,8 +183,18 @@ def index_state(index: str, *, timeout: float = 20.0) -> IndexState:
)
def index_files(index: str, version: str, *, timeout: float = 20.0) -> list[str] | None:
"""Filenames published for ``version`` on ``index``, or ``None`` if absent."""
def index_files(
index: str, version: str, *, timeout: float = 20.0
) -> tuple[list[str] | None, str | None]:
"""Filenames published for ``version``, as ``(filenames, error)``.
``(None, None)`` - the index does not serve this version (yet)
``(None, "why")`` - the lookup itself failed, which is NOT evidence of absence
``([...], None)`` - the version is published with those files
Keeping those apart matters: reporting a network hiccup as "not published
yet" is how a successful upload gets mistaken for a failure.
"""
url = INDEX_JSON[index]
request = urllib.request.Request(
url, headers={"User-Agent": "orisys-release-publish", "Accept": "application/json"}
@@ -191,12 +202,77 @@ def index_files(index: str, version: str, *, timeout: float = 20.0) -> list[str]
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
payload = json.load(response)
except Exception:
return None
except urllib.error.HTTPError as exc:
if exc.code == 404:
return None, None
return None, f"{index} returned HTTP {exc.code}"
except Exception as exc:
return None, f"{index} lookup failed: {type(exc).__name__}: {exc}"
files = (payload.get("releases") or {}).get(version)
if not isinstance(files, list):
return None
return sorted(item.get("filename", "") for item in files)
if not isinstance(files, list) or not files:
return None, None
return sorted(str(item.get("filename", "")) for item in files), None
def wait_for_release(
index: str,
version: str,
*,
attempts: int = 6,
delay: float = 5.0,
sleep=None,
timeout: float = 20.0,
) -> tuple[list[str] | None, str | None]:
"""Poll until the index serves ``version`` (uploads are not instantly visible).
TestPyPI in particular lags the upload response by seconds. Returns the last
``(filenames, error)`` observed.
"""
sleep = sleep or time.sleep
last_error: str | None = None
for attempt in range(1, max(1, attempts) + 1):
filenames, error = index_files(index, version, timeout=timeout)
if filenames is not None:
return filenames, None
last_error = error
if attempt < attempts:
print(
f"[info] {index} does not serve {version} yet "
f"({'lookup failed' if error else 'indexing'}), retry {attempt}/{attempts - 1}",
flush=True,
)
sleep(delay)
return None, last_error
def existing_release_state(index: str, version: str, expected: set[str]) -> str:
"""``"satisfied"`` | ``"conflict"`` | ``"absent"`` for this index and version.
An index version can never be replaced, so "already published with exactly
these files" is the only successful terminal state; treating it as an error
would make a re-run after a partial failure look like a new failure. A
different file set is a genuine conflict and must stop the run.
"""
published, error = index_files(index, version)
if published is None:
if error:
print(f"[warn] could not check {index} for an existing release: {error}")
return "absent"
if set(published) == expected:
print(
f"[ok ] {index} already serves orisys {version} with all "
f"{len(published)} file(s)"
)
return "satisfied"
extra = sorted(set(published) - expected)
missing = sorted(expected - set(published))
print(
f"[FAIL] {index} already serves orisys {version} with a different file set"
+ (f"; unexpected: {extra}" if extra else "")
+ (f"; missing: {missing}" if missing else "")
)
return "conflict"
def _upload(wheels: list[Path], repository: str | None) -> int:
@@ -288,6 +364,16 @@ def publish(options: Options) -> int:
print("[ok ] twine check passed")
# --- 3. phase plan -------------------------------------------------
expected = {wheel.name for wheel in wheels}
remaining: list[str] = []
for index in phases:
state = existing_release_state(index, version, expected)
if state == "conflict":
return 1
if state == "absent":
remaining.append(index)
phases = remaining
print()
for index in phases:
state = index_state(index)
@@ -303,6 +389,11 @@ def publish(options: Options) -> int:
return 1
print(f"[ok ] {index}: {version} not present (published: {', '.join(state.versions) or 'none'})")
if not phases:
print()
print("=== release pypi: every requested index already has this release ===")
return 0
print()
for index in phases:
print(f"[plan] upload {len(wheels)} wheel(s) to {index}")
@@ -324,14 +415,18 @@ def publish(options: Options) -> int:
return 1
print(f"[ok ] uploaded to {index}")
published = index_files(index, version)
published, error = wait_for_release(index, version)
if published is None:
# The upload itself succeeded, so this is a confirmation problem, not
# a publishing failure. Say so plainly and do not imply data loss.
reason = f"lookup kept failing ({error})" if error else "the index has not indexed it yet"
print(
f"[FAIL] {index} does not serve orisys {version} yet; "
"the upload may still be indexing, verify manually"
f"[warn] uploaded, but could not confirm it on {index} yet: {reason}.\n"
f" check https://{'test.' if index == 'testpypi' else ''}pypi.org/project/orisys/{version}/"
"\n re-running this command is safe: an already-published version is skipped"
)
return 1
expected = {wheel.name for wheel in wheels}
print(f"\n=== release pypi: {index} upload SUCCEEDED (verification inconclusive) ===")
return 0
if set(published) != expected:
print(
f"[FAIL] {index} file set mismatch:\n expected {sorted(expected)}\n"
@@ -342,8 +437,12 @@ def publish(options: Options) -> int:
if options.verify_install:
if _fetch_back(index, version, wheels[0]) != 0:
print(f"[FAIL] {index} artifact is not fetchable")
return 1
print(
f"[warn] {index} artifact is not fetchable yet; the upload is "
"published, so this is a propagation delay rather than a failure"
)
else:
print(f"[ok ] fetched the published artifact back from {index}")
if phases == ["testpypi"]:
print()