Use Simple Repository API for fetching latest version (#6704)

redbot/core/_events.py

@@ -11,6 +11,8 @@ import aiohttp
 import discord
 import importlib.metadata
 from packaging.requirements import Requirement
+from packaging.specifiers import SpecifierSet
+from packaging.version import Version
 from redbot.core import data_manager
 
 from redbot.core.bot import ExitCodes
@@ -19,14 +21,13 @@ from redbot.core.i18n import (
     Translator,
     set_contextual_locales_from_guild,
 )
-from .. import __version__ as red_version, version_info as red_version_info
+from .. import __version__ as red_version
 from . import commands
 from ._config import get_latest_confs
 from .utils._internal_utils import (
     fuzzy_command_search,
     format_fuzzy_results,
-    expected_version,
-    fetch_latest_red_version_info,
+    fetch_latest_red_version,
     send_to_owners_with_prefix_replaced,
 )
 from .utils.chat_formatting import inline, format_perms_list
@@ -52,7 +53,7 @@ ______         _           ______ _                       _  ______       _
 _ = Translator(__name__, __file__)
 
 
-def get_outdated_red_messages(pypi_version: str, py_version_req: str) -> Tuple[str, str]:
+def get_outdated_red_messages(pypi_version: str, requires_python: SpecifierSet) -> Tuple[str, str]:
     outdated_red_message = _(
         "Your Red instance is out of date! {} is the current version, however you are using {}!"
     ).format(pypi_version, red_version)
@@ -61,7 +62,7 @@ def get_outdated_red_messages(pypi_version: str, py_version_req: str) -> Tuple[s
         f"[red]!!![/red]Version [cyan]{pypi_version}[/] is available, "
         f"but you're using [cyan]{red_version}[/][red]!!![/red]"
     )
-    current_python = platform.python_version()
+    current_python = Version(platform.python_version())
     extra_update = _(
         "\n\nWhile the following command should work in most scenarios as it is "
         "based on your current OS, environment, and Python version, "
@@ -70,14 +71,14 @@ def get_outdated_red_messages(pypi_version: str, py_version_req: str) -> Tuple[s
         "needs to be done during the update.**"
     ).format(docs="https://docs.discord.red/en/stable/update_red.html")
 
-    if not expected_version(current_python, py_version_req):
+    if current_python not in requires_python:
         extra_update += _(
             "\n\nYou have Python `{py_version}` and this update "
             "requires `{req_py}`; you cannot simply run the update command.\n\n"
             "You will need to follow the update instructions in our docs above, "
             "if you still need help updating after following the docs go to our "
             "#support channel in <https://discord.gg/red>"
-        ).format(py_version=current_python, req_py=py_version_req)
+        ).format(py_version=current_python, req_py=requires_python)
         outdated_red_message += extra_update
         return outdated_red_message, rich_outdated_message
 
@@ -176,7 +177,7 @@ def init_events(bot, cli_flags):
         if bot.intents.members:  # Lets avoid 0 Unique Users
             table_counts.add_row("Unique Users", str(users))
 
-        fetch_version_task = asyncio.create_task(fetch_latest_red_version_info())
+        fetch_version_task = asyncio.create_task(fetch_latest_red_version())
         log.info("Fetching information about latest Red version...")
         try:
             await asyncio.wait_for(asyncio.shield(fetch_version_task), timeout=5)
@@ -214,16 +215,16 @@ def init_events(bot, cli_flags):
         bot._red_ready.set()
 
         try:
-            pypi_version, py_version_req = await fetch_version_task
+            latest = await fetch_version_task
         except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
             log.error("Failed to fetch latest version information from PyPI.", exc_info=exc)
         except (KeyError, ValueError) as exc:
             log.error("Failed to parse version metadata received from PyPI.", exc_info=exc)
         else:
-            outdated = pypi_version and pypi_version > red_version_info
+            outdated = latest.version > Version(red_version)
             if outdated:
                 outdated_red_message, rich_outdated_message = get_outdated_red_messages(
-                    pypi_version, py_version_req
+                    latest.version, latest.requires_python
                 )
                 rich_console.print(rich_outdated_message)
                 await send_to_owners_with_prefix_replaced(bot, outdated_red_message)

redbot/core/core_commands.py

@@ -38,6 +38,7 @@ from typing import (
 
 import aiohttp
 import discord
+from packaging.version import Version
 from redbot.core.data_manager import storage_type
 
 from . import (
@@ -53,7 +54,7 @@ from . import (
 )
 from ._diagnoser import IssueDiagnoser
 from .utils import AsyncIter, can_user_send_messages_in
-from .utils._internal_utils import fetch_latest_red_version_info
+from .utils._internal_utils import fetch_latest_red_version
 from .utils.predicates import MessagePredicate
 from .utils.chat_formatting import (
     box,
@@ -422,11 +423,13 @@ class Core(commands.commands._RuleDropper, commands.Cog, CoreLogic):
         custom_info = await self.bot._config.custom_info()
 
         try:
-            pypi_version, __ = await fetch_latest_red_version_info()
+            latest = await fetch_latest_red_version()
         except (aiohttp.ClientError, TimeoutError) as exc:
             log.error("Failed to fetch latest version information from PyPI.", exc_info=exc)
             pypi_version = None
-        outdated = pypi_version and pypi_version > red_version_info
+        else:
+            pypi_version = latest.version
+        outdated = pypi_version and pypi_version > Version(__version__)
 
         if embed_links:
             dpy_version = "[{}]({})".format(discord.__version__, dpy_repo)

redbot/core/utils/_internal_utils.py

@@ -16,16 +16,19 @@ from io import BytesIO
 from pathlib import Path
 from tarfile import TarInfo
 from typing import (
+    Any,
     AsyncIterable,
     AsyncIterator,
     Awaitable,
     Callable,
+    Dict,
     Generator,
     Iterable,
     Iterator,
     List,
     Optional,
     Union,
+    TypedDict,
     TypeVar,
     TypedDict,
     TYPE_CHECKING,
@@ -35,14 +38,19 @@ from typing import (
 
 import aiohttp
 import discord
-from packaging.requirements import Requirement
+import yarl
+from packaging.metadata import Metadata
+from packaging.specifiers import SpecifierSet
+from packaging.utils import parse_sdist_filename
+from packaging.version import Version
 import rapidfuzz
 import rich.progress
 from rich.console import Console
 from rich.text import Text
 from red_commons.logging import VERBOSE, TRACE
+from typing_extensions import NotRequired, Self
 
-from redbot import VersionInfo
+from redbot import __version__
 from redbot.core import data_manager
 from redbot.core.utils.chat_formatting import box
 
@@ -59,8 +67,10 @@ __all__ = (
     "create_backup",
     "send_to_owners_with_preprocessor",
     "send_to_owners_with_prefix_replaced",
-    "expected_version",
-    "fetch_latest_red_version_info",
+    "ReleaseFile",
+    "AvailableVersion",
+    "fetch_available_red_versions",
+    "fetch_latest_red_version",
     "deprecated_removed",
     "RichIndefiniteBarColumn",
     "RichSpeedColumn",
@@ -70,6 +80,14 @@ __all__ = (
 
 _T = TypeVar("_T")
 
+# I guess there's nothing in allowing people to use an alternative index.
+_SIMPLE_API_URL = os.getenv("RED_SIMPLE_API_URL") or "https://pypi.org/simple/"
+# This variable should only be used for debugging purposes (hence why it starts with `_`).
+# You can debug the behavior by e.g. creating a "Red-DiscordBot.json" file,
+# starting a server with `python -m http.server` and starting Red with the following env vars:
+# RED_SIMPLE_API_URL=http://localhost:8000 _RED_SIMPLE_API_ENDPOINT_PATH=Red-DiscordBot.json
+_SIMPLE_API_ENDPOINT_PATH = os.getenv("_RED_SIMPLE_API_ENDPOINT_PATH") or "Red-DiscordBot"
+
 
 def safe_delete(pth: Path):
     if pth.exists():
@@ -378,14 +396,100 @@ async def send_to_owners_with_prefix_replaced(bot: Red, content: str, **kwargs):
     await send_to_owners_with_preprocessor(bot, content, content_preprocessor=preprocessor)
 
 
-def expected_version(current: str, expected: str) -> bool:
-    # Requirement needs a regular requirement string, so "x" serves as requirement's name here
-    return Requirement(f"x{expected}").specifier.contains(current, prereleases=True)
+# gotta use functional TypedDict syntax due to hyphens in keys
+ReleaseFile = TypedDict(
+    "ReleaseFile",
+    {
+        "filename": str,
+        "url": str,
+        "hashes": Dict[str, str],
+        "requires-python": NotRequired[str],
+        "core-metadata": NotRequired[Union[bool, Dict[str, str]]],
+        "yanked": bool,
+        "size": int,
+        "upload-time": NotRequired[str],
+        "provenance": NotRequired[Optional[str]],
+    },
+)
 
 
-async def fetch_latest_red_version_info() -> Tuple[VersionInfo, Optional[str]]:
+class AvailableVersion:
+    def __init__(self, version: Version, files: Dict[str, ReleaseFile]) -> None:
+        self.version = version
+        self.files = files
+        required_pythons = {f.get("requires-python") or "" for f in files.values()}
+        if len(required_pythons) > 1:
+            raise ValueError("found multiple files with different Requires-Python values")
+        self.requires_python = SpecifierSet(required_pythons.pop())
+
+    @classmethod
+    def from_json_dict(cls, data: Dict[str, Any]) -> Self:
+        ret = cls(Version(data["version"]), data["files"])
+        if str(ret.requires_python) != data["requires_python"]:
+            raise ValueError("requires_python key in given data is inconsistent with files")
+        return ret
+
+    def to_json_dict(self) -> Dict[str, Any]:
+        return {
+            "version": str(self.version),
+            "requires_python": str(self.requires_python),
+            "files": self.files,
+        }
+
+    async def fetch_core_metadata(self) -> Metadata:
+        for release_file in self.files.values():
+            core_metadata_hashes = release_file.get("core-metadata", False)
+            if core_metadata_hashes is False:
+                continue
+            async with aiohttp.ClientSession() as session:
+                async with session.get(f"{release_file['url']}.metadata") as resp:
+                    return Metadata.from_email(await resp.read(), validate=False)
+        raise TypeError("Could not find core metadata for any of the release files.")
+
+    def __eq__(self, other: Any) -> bool:
+        if isinstance(other, self.__class__):
+            return self.version == other.version
+        return NotImplemented
+
+    def __ne__(self, other: Any) -> bool:
+        if isinstance(other, self.__class__):
+            return self.version != other.version
+        return NotImplemented
+
+    def __lt__(self, other: Any) -> bool:
+        if isinstance(other, self.__class__):
+            return self.version < other.version
+        return NotImplemented
+
+    def __le__(self, other: Any) -> bool:
+        if isinstance(other, self.__class__):
+            return self.version <= other.version
+        return NotImplemented
+
+    def __gt__(self, other: Any) -> bool:
+        if isinstance(other, self.__class__):
+            return self.version > other.version
+        return NotImplemented
+
+    def __ge__(self, other: Any) -> bool:
+        if isinstance(other, self.__class__):
+            return self.version >= other.version
+        return NotImplemented
+
+
+async def fetch_available_red_versions(
+    *, include_prereleases: Optional[bool] = None
+) -> List[AvailableVersion]:
     """
-    Fetch information about latest Red release on PyPI.
+    Fetch information about Red releases available on PyPI,
+    sorted by version (latest first).
+
+    Parameters
+    ----------
+    include_prereleases : bool, optional
+        Whether the pre-releases should be included in the list.
+        If ``None`` (the default), the pre-releases will only be included,
+        if the currently running Red version is considered a pre-release.
 
     Raises
     ------
@@ -394,18 +498,93 @@ async def fetch_latest_red_version_info() -> Tuple[VersionInfo, Optional[str]]:
     TimeoutError
         The request to PyPI timed out.
     ValueError
-        An invalid version string was returned in PyPI metadata.
+        Some part of the response was considered invalid.
+        This includes issues such as incorrect response content type,
+        invalid version strings, inability to find files for a release,
+        and mismatching Requires-Python values.
     KeyError
         The PyPI metadata is missing some of the required information.
     """
+    if include_prereleases is None:
+        include_prereleases = Version(__version__).is_prerelease
+    expected_content_type = "application/vnd.pypi.simple.v1+json"
     async with aiohttp.ClientSession() as session:
-        async with session.get("https://pypi.org/pypi/Red-DiscordBot/json") as r:
-            data = await r.json()
+        async with session.get(
+            yarl.URL(_SIMPLE_API_URL) / _SIMPLE_API_ENDPOINT_PATH,
+            headers={"Accept": expected_content_type},
+        ) as resp:
+            data = await resp.json()
+            content_type = resp.headers["Content-Type"]
+            if not (
+                content_type.startswith(expected_content_type)
+                or (
+                    content_type.startswith("application/json")
+                    and data["meta"]["api-version"].startswith("1.")
+                )
+            ):
+                raise ValueError("got unexpected response from Simple Repository API")
 
-    release = VersionInfo.from_str(data["info"]["version"])
-    required_python = data["info"]["requires_python"]
+    files: Dict[Version, Dict[str, ReleaseFile]] = {}
+    f: ReleaseFile
+    for f in data["files"]:
+        if f.get("yanked"):
+            continue
+        filename = f["filename"]
+        if filename.endswith((".tar.gz", ".zip")):
+            _, version = parse_sdist_filename(filename)
+        elif filename.endswith(".whl"):
+            # https://packaging.python.org/en/latest/specifications/binary-distribution-format/#file-name-convention
+            _, raw_version, _ = filename.split("-", 2)
+            version = Version(raw_version)
+        else:
+            continue
+        if version.is_prerelease and not include_prereleases:
+            continue
+        version_files = files.setdefault(version, {})
+        version_files[f["filename"]] = f
 
-    return release, required_python
+    if not files:
+        raise ValueError("could not find any files")
+
+    available_versions = [
+        AvailableVersion(version, version_files) for version, version_files in files.items()
+    ]
+    available_versions.sort(reverse=True)
+
+    return available_versions
+
+
+async def fetch_latest_red_version(
+    *, include_prereleases: Optional[bool] = None
+) -> AvailableVersion:
+    """
+    Fetch information about latest Red release on PyPI.
+
+    Parameters
+    ----------
+    include_prereleases : bool, optional
+        Whether the pre-releases should be considered when finding the latest version.
+        If ``None`` (the default), the pre-releases will only be considered,
+        if the currently running Red version is considered a pre-release.
+
+    Raises
+    ------
+    aiohttp.ClientError
+        An error occurred during request to PyPI.
+    TimeoutError
+        The request to PyPI timed out.
+    ValueError
+        Some part of the response was considered invalid.
+        This includes issues such as incorrect response content type,
+        invalid version strings, inability to find files for a release,
+        and mismatching Requires-Python values.
+    KeyError
+        The PyPI metadata is missing some of the required information.
+    """
+    available_versions = await fetch_available_red_versions(
+        include_prereleases=include_prereleases
+    )
+    return available_versions[0]
 
 
 def deprecated_removed(