Rip out Downloader's non-UI functionality into private core API (#6706)

This commit is contained in:
Jakub Kuczys
2026-03-29 22:25:04 +02:00
committed by GitHub
parent e2acec0862
commit ee1db01a2f
20 changed files with 1120 additions and 896 deletions
-1
View File
@@ -6,4 +6,3 @@ from .downloader import Downloader
async def setup(bot: Red) -> None:
cog = Downloader(bot)
await bot.add_cog(cog)
cog.create_init_task()
+15 -7
View File
@@ -1,7 +1,8 @@
import discord
from redbot.core import commands
from redbot.core import _downloader, commands
from redbot.core.i18n import Translator
from .installable import InstalledModule
from redbot.core._downloader.installable import InstalledModule
from redbot.core._downloader.repo_manager import Repo as _Repo
_ = Translator("Koala", __file__)
@@ -9,14 +10,21 @@ _ = Translator("Koala", __file__)
class InstalledCog(InstalledModule):
@classmethod
async def convert(cls, ctx: commands.Context, arg: str) -> InstalledModule:
downloader = ctx.bot.get_cog("Downloader")
if downloader is None:
raise commands.CommandError(_("No Downloader cog found."))
cog = discord.utils.get(await downloader.installed_cogs(), name=arg)
cog = discord.utils.get(await _downloader.installed_cogs(), name=arg)
if cog is None:
raise commands.BadArgument(
_("Cog `{cog_name}` is not installed.").format(cog_name=arg)
)
return cog
class Repo(_Repo):
@classmethod
async def convert(cls, ctx: commands.Context, argument: str) -> _Repo:
poss_repo = _downloader._repo_manager.get_repo(argument)
if poss_repo is None:
raise commands.BadArgument(
_('Repo by the name "{repo_name}" does not exist.').format(repo_name=argument)
)
return poss_repo
File diff suppressed because it is too large Load Diff
-165
View File
@@ -1,165 +0,0 @@
from __future__ import annotations
from typing import List, TYPE_CHECKING
if TYPE_CHECKING:
from .repo_manager import Candidate
__all__ = [
"DownloaderException",
"GitException",
"InvalidRepoName",
"CopyingError",
"ExistingGitRepo",
"MissingGitRepo",
"AuthenticationError",
"CloningError",
"CurrentHashError",
"HardResetError",
"UpdateError",
"GitDiffError",
"NoRemoteURL",
"UnknownRevision",
"AmbiguousRevision",
"PipError",
]
class DownloaderException(Exception):
"""
Base class for Downloader exceptions.
"""
pass
class GitException(DownloaderException):
"""
Generic class for git exceptions.
"""
def __init__(self, message: str, git_command: str) -> None:
self.git_command = git_command
super().__init__(f"Git command failed: {git_command}\nError message: {message}")
class InvalidRepoName(DownloaderException):
"""
Throw when a repo name is invalid. Check
the message for a more detailed reason.
"""
pass
class CopyingError(DownloaderException):
"""
Throw when there was an issue
during copying of module's files.
"""
pass
class ExistingGitRepo(DownloaderException):
"""
Thrown when trying to clone into a folder where a
git repo already exists.
"""
pass
class MissingGitRepo(DownloaderException):
"""
Thrown when a git repo is expected to exist but
does not.
"""
pass
class AuthenticationError(GitException):
"""
Thrown when git failed to authenticate with
the server
"""
pass
class CloningError(GitException):
"""
Thrown when git clone returns a non zero exit code.
"""
pass
class CurrentHashError(GitException):
"""
Thrown when git returns a non zero exit code attempting
to determine the current commit hash.
"""
pass
class HardResetError(GitException):
"""
Thrown when there is an issue trying to execute a hard reset
(usually prior to a repo update).
"""
pass
class UpdateError(GitException):
"""
Thrown when git pull returns a non zero error code.
"""
pass
class GitDiffError(GitException):
"""
Thrown when a git diff fails.
"""
pass
class NoRemoteURL(GitException):
"""
Thrown when no remote URL exists for a repo.
"""
pass
class UnknownRevision(GitException):
"""
Thrown when specified revision cannot be found.
"""
pass
class AmbiguousRevision(GitException):
"""
Thrown when specified revision is ambiguous.
"""
def __init__(self, message: str, git_command: str, candidates: List[Candidate]) -> None:
super().__init__(message, git_command)
self.candidates = candidates
class PipError(DownloaderException):
"""
Thrown when pip returns a non-zero return code.
"""
pass
-231
View File
@@ -1,231 +0,0 @@
from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING, Any, Callable, Dict, Tuple, Union, cast
from redbot import VersionInfo, version_info as red_version_info
from . import installable
from .log import log
if TYPE_CHECKING:
from .json_mixins import RepoJSONMixin
__all__ = ("REPO_SCHEMA", "INSTALLABLE_SCHEMA", "update_mixin")
class UseDefault:
"""To be used as sentinel."""
# sentinel value
USE_DEFAULT = UseDefault()
def ensure_tuple_of_str(
info_file: Path, key_name: str, value: Union[Any, UseDefault]
) -> Tuple[str, ...]:
default: Tuple[str, ...] = ()
if value is USE_DEFAULT:
return default
if not isinstance(value, list):
log.warning(
"Invalid value of '%s' key (expected list, got %s)"
" in JSON information file at path: %s",
key_name,
type(value).__name__,
info_file,
)
return default
for item in value:
if not isinstance(item, str):
log.warning(
"Invalid item in '%s' list (expected str, got %s)"
" in JSON information file at path: %s",
key_name,
type(item).__name__,
info_file,
)
return default
return tuple(value)
def ensure_str(info_file: Path, key_name: str, value: Union[Any, UseDefault]) -> str:
default = ""
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
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 ensure_python_version_info(
info_file: Path, key_name: str, value: Union[Any, UseDefault]
) -> Tuple[int, int, int]:
default = (3, 5, 1)
if value is USE_DEFAULT:
return default
if not isinstance(value, list):
log.warning(
"Invalid value of '%s' key (expected list, got %s)"
" in JSON information file at path: %s",
key_name,
type(value).__name__,
info_file,
)
return default
count = len(value)
if count != 3:
log.warning(
"Invalid value of '%s' key (expected list with 3 items, got %s items)"
" in JSON information file at path: %s",
key_name,
count,
info_file,
)
return default
for item in value:
if not isinstance(item, int):
log.warning(
"Invalid item in '%s' list (expected int, got %s)"
" in JSON information file at path: %s",
key_name,
type(item).__name__,
info_file,
)
return default
return cast(Tuple[int, int, int], tuple(value))
def ensure_bool(
info_file: Path, key_name: str, value: Union[Any, UseDefault], *, default: bool = False
) -> bool:
if value is USE_DEFAULT:
return default
if not isinstance(value, bool):
log.warning(
"Invalid value of '%s' key (expected bool, got %s)"
" in JSON information file at path: %s",
key_name,
type(value).__name__,
info_file,
)
return default
return value
def ensure_required_cogs_mapping(
info_file: Path, key_name: str, value: Union[Any, UseDefault]
) -> Dict[str, str]:
default: Dict[str, str] = {}
if value is USE_DEFAULT:
return default
if not isinstance(value, dict):
log.warning(
"Invalid value of '%s' key (expected dict, got %s)"
" in JSON information file at path: %s",
key_name,
type(value).__name__,
info_file,
)
return default
# keys in json dicts are always strings
for item in value.values():
if not isinstance(item, str):
log.warning(
"Invalid item in '%s' dict (expected str, got %s)"
" in JSON information file at path: %s",
key_name,
type(item).__name__,
info_file,
)
return default
return value
def ensure_installable_type(
info_file: Path, key_name: str, value: Union[Any, UseDefault]
) -> installable.InstallableType:
default = installable.InstallableType.COG
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 # NOTE: old behavior was to use InstallableType.UNKNOWN
if value in ("", "COG"):
return installable.InstallableType.COG
if value == "SHARED_LIBRARY":
return installable.InstallableType.SHARED_LIBRARY
return installable.InstallableType.UNKNOWN
EnsureCallable = Callable[[Path, str, Union[Any, UseDefault]], Any]
SchemaType = Dict[str, EnsureCallable]
REPO_SCHEMA: SchemaType = {
"author": ensure_tuple_of_str,
"description": ensure_str,
"install_msg": ensure_str,
"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,
"hidden": ensure_bool,
"disabled": ensure_bool,
"required_cogs": ensure_required_cogs_mapping,
"requirements": ensure_tuple_of_str,
"tags": ensure_tuple_of_str,
"type": ensure_installable_type,
"end_user_data_statement": ensure_str,
}
def update_mixin(repo_or_installable: RepoJSONMixin, schema: SchemaType) -> None:
info = repo_or_installable._info
info_file = repo_or_installable._info_file
for key, callback in schema.items():
setattr(repo_or_installable, key, callback(info_file, key, info.get(key, USE_DEFAULT)))
-206
View File
@@ -1,206 +0,0 @@
from __future__ import annotations
import functools
import shutil
from enum import IntEnum
from pathlib import Path
from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, Tuple, Union, cast
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
class InstallableType(IntEnum):
# using IntEnum, because hot-reload breaks its identity
UNKNOWN = 0
COG = 1
SHARED_LIBRARY = 2
class Installable(RepoJSONMixin):
"""Base class for anything the Downloader cog can install.
- Modules
- Repo Libraries
- Other stuff?
The attributes of this class will mostly come from the installation's
info.json.
Attributes
----------
repo_name : `str`
Name of the repository which this package belongs to.
repo : Repo, optional
Repo object of the Installable, if repo is missing this will be `None`
commit : `str`, optional
Installable's commit. This is not the same as ``repo.commit``
author : `tuple` of `str`
Name(s) of the author(s).
end_user_data_statement : `str`
End user data statement of the module.
min_bot_version : `VersionInfo`
The minimum bot version required for this Installable.
max_bot_version : `VersionInfo`
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`
The minimum python version required for this cog.
hidden : `bool`
Whether or not this cog will be hidden from the user when they use
`Downloader`'s commands.
required_cogs : `dict`
In the form :code:`{cog_name : repo_url}`, these are cogs which are
required for this installation.
requirements : `tuple` of `str`
Required libraries for this installation.
tags : `tuple` of `str`
List of tags to assist in searching.
type : `int`
The type of this installation, as specified by
:class:`InstallationType`.
"""
def __init__(self, location: Path, repo: Optional[Repo] = None, commit: str = ""):
"""Base installable initializer.
Parameters
----------
location : pathlib.Path
Location (file or folder) to the installable.
repo : Repo, optional
Repo object of the Installable, if repo is missing this will be `None`
commit : str
Installable's commit. This is not the same as ``repo.commit``
"""
self._location = location
self.repo = repo
self.repo_name = self._location.parent.name
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.hidden: bool
self.disabled: bool
self.required_cogs: Dict[str, str] # Cog name -> repo URL
self.requirements: Tuple[str, ...]
self.tags: Tuple[str, ...]
self.type: InstallableType
super().__init__(location)
def __eq__(self, other: Any) -> bool:
# noinspection PyProtectedMember
return self._location == other._location
def __hash__(self) -> int:
return hash(self._location)
@property
def name(self) -> str:
"""`str` : The name of this package."""
return self._location.stem
async def copy_to(self, target_dir: Path) -> bool:
"""
Copies this cog/shared_lib to the given directory. This
will overwrite any files in the target directory.
:param pathlib.Path target_dir: The installation directory to install to.
:return: Status of installation
:rtype: bool
"""
copy_func: Callable[..., Any]
if self._location.is_file():
copy_func = shutil.copy2
else:
copy_func = functools.partial(shutil.copytree, dirs_exist_ok=True)
# noinspection PyBroadException
try:
copy_func(src=str(self._location), dst=str(target_dir / self._location.name))
except: # noqa: E722
log.exception("Error occurred when copying path: %s", self._location)
return False
return True
def _read_info_file(self) -> None:
super()._read_info_file()
update_mixin(self, INSTALLABLE_SCHEMA)
if self.type == InstallableType.SHARED_LIBRARY:
self.hidden = True
class InstalledModule(Installable):
"""Base class for installed modules,
this is basically instance of installed `Installable`
used by Downloader.
Attributes
----------
pinned : `bool`
Whether or not this cog is pinned, always `False` if module is not a cog.
"""
def __init__(
self,
location: Path,
repo: Optional[Repo] = None,
commit: str = "",
pinned: bool = False,
json_repo_name: str = "",
):
super().__init__(location=location, repo=repo, commit=commit)
self.pinned: bool = pinned if self.type == InstallableType.COG else False
# this is here so that Downloader could use real repo name instead of "MISSING_REPO"
self._json_repo_name = json_repo_name
def to_json(self) -> Dict[str, Union[str, bool]]:
module_json: Dict[str, Union[str, bool]] = {
"repo_name": self.repo_name,
"module_name": self.name,
"commit": self.commit,
}
if self.type == InstallableType.COG:
module_json["pinned"] = self.pinned
return module_json
@classmethod
def from_json(
cls, data: Dict[str, Union[str, bool]], repo_mgr: RepoManager
) -> InstalledModule:
repo_name = cast(str, data["repo_name"])
cog_name = cast(str, data["module_name"])
commit = cast(str, data.get("commit", ""))
pinned = cast(bool, data.get("pinned", False))
# TypedDict, where are you :/
repo = repo_mgr.get_repo(repo_name)
if repo is not None:
repo_folder = repo.folder_path
else:
repo_folder = repo_mgr.repos_folder / "MISSING_REPO"
location = repo_folder / cog_name
return cls(
location=location, repo=repo, commit=commit, pinned=pinned, json_repo_name=repo_name
)
@classmethod
def from_installable(cls, module: Installable, *, pinned: bool = False) -> InstalledModule:
return cls(
location=module._location, repo=module.repo, commit=module.commit, pinned=pinned
)
-47
View File
@@ -1,47 +0,0 @@
import json
from pathlib import Path
from typing import Any, Dict, Tuple
from .info_schemas import REPO_SCHEMA, update_mixin
from .log import log
class RepoJSONMixin:
INFO_FILE_NAME = "info.json"
def __init__(self, repo_folder: Path):
self._repo_folder = repo_folder
self.author: Tuple[str, ...]
self.install_msg: str
self.short: str
self.description: str
self._info_file = repo_folder / self.INFO_FILE_NAME
self._info: Dict[str, Any]
self._read_info_file()
def _read_info_file(self) -> None:
if self._info_file.exists():
try:
with self._info_file.open(encoding="utf-8") as f:
info = json.load(f)
except json.JSONDecodeError as e:
log.error(
"Invalid JSON information file at path: %s\nError: %s", self._info_file, str(e)
)
info = {}
else:
info = {}
if not isinstance(info, dict):
log.warning(
"Invalid top-level structure (expected dict, got %s)"
" in JSON information file at path: %s",
type(info).__name__,
self._info_file,
)
info = {}
self._info = info
update_mixin(self, REPO_SCHEMA)
File diff suppressed because it is too large Load Diff