Add Environment abstraction to internal Downloader API (#6710)

This commit is contained in:
Jakub Kuczys
2026-05-14 00:09:45 +02:00
committed by GitHub
parent 13f45f69ac
commit 899f24ceca
6 changed files with 120 additions and 67 deletions
+4 -5
View File
@@ -4,7 +4,8 @@ import re
from typing import Tuple, Iterable, Collection, Optional, Set, List
import discord
from redbot.core import _downloader, commands, version_info as red_version_info
from redbot import __version__
from redbot.core import _downloader, commands
from redbot.core._downloader import errors
from redbot.core._downloader.installable import InstalledModule
from redbot.core.bot import Red
@@ -956,9 +957,7 @@ class Downloader(commands.Cog):
) + humanize_list(
[
inline(cog.name)
+ _(" (Minimum: {min_version})").format(
min_version=".".join([str(n) for n in cog.min_python_version])
)
+ _(" (Minimum: {min_version})").format(min_version=cog.min_python_version)
for cog in update_check_result.incompatible_python_version
]
)
@@ -973,7 +972,7 @@ class Downloader(commands.Cog):
"\nThis cog requires different Red version than you currently "
"have ({current_version}): "
)
).format(current_version=red_version_info) + humanize_list(
).format(current_version=__version__) + humanize_list(
[
inline(cog.name)
+ _(" (Minimum: {min_version}").format(min_version=cog.min_bot_version)
+58 -16
View File
@@ -12,6 +12,7 @@ from __future__ import annotations
import contextlib
import dataclasses
import functools
import json
import os
import shutil
@@ -35,7 +36,11 @@ from typing import (
)
import discord
from redbot.core import commands, Config, version_info as red_version_info
from packaging.version import Version
from typing_extensions import Self
from redbot import __version__
from redbot.core import commands, Config
from redbot.core._cog_manager import CogManager
from redbot.core.data_manager import cog_data_path
from redbot.core.utils._internal_utils import detailed_progress
@@ -554,8 +559,34 @@ async def reinstall_requirements() -> Tuple[Tuple[str, ...], Tuple[Installable,
return failed_reqs, tuple(all_failed_libs)
# TODO: make kw_only
@dataclasses.dataclass(frozen=True)
class Environment:
"""
Environment to check version bounds against.
Usually the metadata from current environment is used but this allows for alternative uses
such as checking, if the cog will work on a Red version you want to update to.
"""
red_version: Version
python_version: Version
@classmethod
@functools.lru_cache(maxsize=None)
def current(cls) -> Self:
return cls(
red_version=Version(__version__),
python_version=Version(".".join(map(str, sys.version_info[:3]))),
)
async def install_cogs(
repo: Repo, rev: Optional[str], cog_names: Iterable[str]
repo: Repo,
rev: Optional[str],
cog_names: Iterable[str],
*,
env: Environment = Environment.current(),
) -> CogInstallResult:
commit = None
@@ -588,12 +619,12 @@ async def install_cogs(
already_installed.append(cog)
elif discord.utils.get(_installed_cogs, name=cog.name):
name_already_used.append(cog)
elif cog.min_python_version > sys.version_info:
elif cog.min_python_version > env.python_version:
incompatible_python_version.append(cog)
elif cog.min_bot_version > red_version_info or (
elif cog.min_bot_version > env.red_version or (
# max version should be ignored when it's lower than min version
cog.min_bot_version <= cog.max_bot_version
and cog.max_bot_version < red_version_info
and cog.max_bot_version < env.red_version
):
incompatible_bot_version.append(cog)
else:
@@ -653,6 +684,7 @@ async def check_cog_updates(
repos: Optional[Iterable[Repo]] = None,
cogs: Optional[Iterable[InstalledModule]] = None,
update_repos: bool = True,
env: Environment = Environment.current(),
) -> CogUpdateCheckResult:
cogs_to_check, failed_repos = await _get_cogs_to_check(
repos=repos, cogs=cogs, update_repos=update_repos
@@ -663,12 +695,12 @@ async def check_cog_updates(
incompatible_python_version: List[Installable] = []
incompatible_bot_version: List[Installable] = []
for cog in outdated_cogs:
if cog.min_python_version > sys.version_info:
if cog.min_python_version > env.python_version:
incompatible_python_version.append(cog)
elif cog.min_bot_version > red_version_info or (
elif cog.min_bot_version > env.red_version or (
# max version should be ignored when it's lower than min version
cog.min_bot_version <= cog.max_bot_version
and cog.max_bot_version < red_version_info
and cog.max_bot_version < env.red_version
):
incompatible_bot_version.append(cog)
else:
@@ -686,19 +718,26 @@ async def check_cog_updates(
# update given cogs or all cogs
async def update_cogs(
*, cogs: Optional[List[InstalledModule]] = None, repos: Optional[List[Repo]] = None
*,
cogs: Optional[List[InstalledModule]] = None,
repos: Optional[List[Repo]] = None,
env: Environment = Environment.current(),
) -> CogUpdateResult:
if cogs is not None and repos is not None:
raise ValueError("You can specify cogs or repos argument, not both")
cogs_to_check, failed_repos = await _get_cogs_to_check(repos=repos, cogs=cogs)
return await _update_cogs(cogs_to_check, failed_repos=failed_repos)
return await _update_cogs(cogs_to_check, failed_repos=failed_repos, env=env)
# update given cogs or all cogs from the specified repo
# using the specified revision (or latest if not specified)
async def update_repo_cogs(
repo: Repo, cogs: Optional[List[InstalledModule]] = None, *, rev: Optional[str] = None
repo: Repo,
cogs: Optional[List[InstalledModule]] = None,
*,
rev: Optional[str] = None,
env: Environment = Environment.current(),
) -> CogUpdateResult:
try:
await repo.update()
@@ -712,11 +751,14 @@ async def update_repo_cogs(
commit = await repo.get_full_sha1(rev)
async with repo.checkout(commit, exit_to_rev=repo.branch):
cogs_to_check, __ = await _get_cogs_to_check(repos=[repo], cogs=cogs, update_repos=False)
return await _update_cogs(cogs_to_check, failed_repos=())
return await _update_cogs(cogs_to_check, failed_repos=(), env=env)
async def _update_cogs(
cogs_to_check: Set[InstalledModule], *, failed_repos: Sequence[str]
cogs_to_check: Set[InstalledModule],
*,
failed_repos: Sequence[str],
env: Environment = Environment.current(),
) -> CogUpdateResult:
pinned_cogs = {cog for cog in cogs_to_check if cog.pinned}
cogs_to_check -= pinned_cogs
@@ -737,12 +779,12 @@ async def _update_cogs(
outdated_cogs, outdated_libs = await _available_updates(cogs_to_check)
for cog in outdated_cogs:
if cog.min_python_version > sys.version_info:
if cog.min_python_version > env.python_version:
incompatible_python_version.append(cog)
elif cog.min_bot_version > red_version_info or (
elif cog.min_bot_version > env.red_version or (
# max version should be ignored when it's lower than min version
cog.min_bot_version <= cog.max_bot_version
and cog.max_bot_version < red_version_info
and cog.max_bot_version < env.red_version
):
incompatible_bot_version.append(cog)
else:
+41 -35
View File
@@ -1,9 +1,9 @@
from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING, Any, Callable, Dict, Tuple, Union, cast
from typing import TYPE_CHECKING, Any, Callable, Dict, Tuple, Union
from redbot import VersionInfo, version_info as red_version_info
from packaging.version import Version
from . import installable
from .log import log
@@ -67,38 +67,40 @@ def ensure_str(info_file: Path, key_name: str, value: Union[Any, UseDefault]) ->
return value
def ensure_red_version_info(
info_file: Path, key_name: str, value: Union[Any, UseDefault]
) -> VersionInfo:
default = red_version_info
if value is USE_DEFAULT:
return default
if not isinstance(value, str):
log.warning(
"Invalid value of '%s' key (expected str, got %s)"
" in JSON information file at path: %s",
key_name,
type(value).__name__,
info_file,
)
return default
try:
version_info = VersionInfo.from_str(value)
except ValueError:
log.warning(
"Invalid value of '%s' key (given value isn't a valid version string)"
" in JSON information file at path: %s",
key_name,
info_file,
)
return default
return version_info
def create_ensure_red_version(default: Version) -> EnsureCallable:
def ensure_red_version(
info_file: Path, key_name: str, value: Union[Any, UseDefault]
) -> Version:
if value is USE_DEFAULT:
return default
if not isinstance(value, str):
log.warning(
"Invalid value of '%s' key (expected str, got %s)"
" in JSON information file at path: %s",
key_name,
type(value).__name__,
info_file,
)
return default
try:
version_info = Version(value)
except ValueError:
log.warning(
"Invalid value of '%s' key (given value isn't a valid version string)"
" in JSON information file at path: %s",
key_name,
info_file,
)
return default
return version_info
return ensure_red_version
def ensure_python_version_info(
def ensure_python_version(
info_file: Path, key_name: str, value: Union[Any, UseDefault]
) -> Tuple[int, int, int]:
default = (3, 5, 1)
) -> Version:
default = Version("3.5.1")
if value is USE_DEFAULT:
return default
if not isinstance(value, list):
@@ -130,7 +132,7 @@ def ensure_python_version_info(
info_file,
)
return default
return cast(Tuple[int, int, int], tuple(value))
return Version(".".join(map(str, value)))
def ensure_bool(
@@ -211,9 +213,13 @@ REPO_SCHEMA: SchemaType = {
"short": ensure_str,
}
INSTALLABLE_SCHEMA: SchemaType = {
"min_bot_version": ensure_red_version_info,
"max_bot_version": ensure_red_version_info,
"min_python_version": ensure_python_version_info,
"min_bot_version": create_ensure_red_version(Version("0.0.dev0")),
# Using little-known version epoch feature to represent something that,
# for all practical purposes, will be considered higher than any version number
# that we may ever have.
# https://packaging.python.org/en/latest/specifications/version-specifiers/#version-epochs
"max_bot_version": create_ensure_red_version(Version("99999!99999.99999.post99999+hi.mom")),
"min_python_version": ensure_python_version,
"hidden": ensure_bool,
"disabled": ensure_bool,
"required_cogs": ensure_required_cogs_mapping,
+8 -8
View File
@@ -6,12 +6,12 @@ from enum import Enum
from pathlib import Path
from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, Tuple, Union, cast
from packaging.version import Version
from .log import log
from .info_schemas import INSTALLABLE_SCHEMA, update_mixin
from .json_mixins import RepoJSONMixin
from redbot.core import VersionInfo
if TYPE_CHECKING:
from .repo_manager import RepoManager, Repo
@@ -44,12 +44,12 @@ class Installable(RepoJSONMixin):
Name(s) of the author(s).
end_user_data_statement : `str`
End user data statement of the module.
min_bot_version : `VersionInfo`
min_bot_version : `packaging.version.Version`
The minimum bot version required for this Installable.
max_bot_version : `VersionInfo`
max_bot_version : `packaging.version.Version`
The maximum bot version required for this Installable.
Ignored if `min_bot_version` is newer than `max_bot_version`.
min_python_version : `tuple` of `int`
min_python_version : `packaging.version.Version`
The minimum python version required for this cog.
hidden : `bool`
Whether or not this cog will be hidden from the user when they use
@@ -87,9 +87,9 @@ class Installable(RepoJSONMixin):
self.commit = commit
self.end_user_data_statement: str
self.min_bot_version: VersionInfo
self.max_bot_version: VersionInfo
self.min_python_version: Tuple[int, int, int]
self.min_bot_version: Version
self.max_bot_version: Version
self.min_python_version: Version
self.hidden: bool
self.disabled: bool
self.required_cogs: Dict[str, str] # Cog name -> repo URL
+2
View File
@@ -87,6 +87,7 @@ INFO_JSON = {
"author": ("tekulvw",),
"min_bot_version": "3.0.0",
"max_bot_version": "3.0.2",
"min_python_version": [3, 7, 1],
"description": "A long description",
"hidden": False,
"install_msg": "A post-installation message",
@@ -101,6 +102,7 @@ LIBRARY_INFO_JSON = {
"author": ("seputaes",),
"min_bot_version": "3.0.0",
"max_bot_version": "3.0.2",
"min_python_version": [3, 7, 1],
"description": "A long library description",
"hidden": False, # libraries are always hidden, this tests it will be flipped
"install_msg": "A library install message",
+7 -3
View File
@@ -2,10 +2,10 @@ import json
from pathlib import Path
import pytest
from packaging.version import Version
from redbot.pytest.downloader import *
from redbot.core._downloader.installable import Installable, InstallableType
from redbot.core import VersionInfo
def test_process_info_file(installable):
@@ -13,7 +13,9 @@ def test_process_info_file(installable):
if k == "type":
assert installable.type is InstallableType.COG
elif k in ("min_bot_version", "max_bot_version"):
assert getattr(installable, k) == VersionInfo.from_str(v)
assert getattr(installable, k) == Version(v)
elif k == "min_python_version":
assert installable.min_python_version == Version(".".join(map(str, v)))
else:
assert getattr(installable, k) == v
@@ -23,7 +25,9 @@ def test_process_lib_info_file(library_installable):
if k == "type":
assert library_installable.type is InstallableType.SHARED_LIBRARY
elif k in ("min_bot_version", "max_bot_version"):
assert getattr(library_installable, k) == VersionInfo.from_str(v)
assert getattr(library_installable, k) == Version(v)
elif k == "min_python_version":
assert library_installable.min_python_version == Version(".".join(map(str, v)))
elif k == "hidden":
# libraries are always hidden, even if False
assert library_installable.hidden is True