Split public and private data manager APIs into separate modules

.github/labeler.yml

@@ -225,6 +225,7 @@
 "Category: Core - Other Internals":
   # Source
   - redbot/core/_cog_manager.py
+  - redbot/core/_data_manager.py
   - redbot/core/_events.py
   - redbot/core/_global_checks.py
   - redbot/core/_settings_caches.py

redbot/__main__.py

@@ -27,7 +27,7 @@
 from redbot.core.bot import Red, ExitCodes, _NoOwnerSet
 from redbot.core._cli import interactive_config, confirm, parse_cli_flags
 from redbot.setup import get_data_dir, get_name, save_config
-from redbot.core import data_manager, _drivers
+from redbot.core import _data_manager, data_manager, _drivers
 from redbot.core._debuginfo import DebugInfo
 from redbot.core._sharedlibdeprecation import SharedLibImportWarner
 
@@ -42,13 +42,13 @@
 
 
 def _get_instance_names():
-    with data_manager.config_file.open(encoding="utf-8") as fs:
+    with _data_manager.config_file.open(encoding="utf-8") as fs:
         data = json.load(fs)
     return sorted(data.keys())
 
 
 def list_instances():
-    if not data_manager.config_file.exists():
+    if not _data_manager.config_file.exists():
         print(
             "No instances have been configured! Configure one "
             "using `redbot-setup` before trying to run the bot!"
@@ -100,7 +100,7 @@ async def edit_instance(red, cli_flags):
     await _edit_prefix(red, prefix, no_prompt)
     await _edit_owner(red, owner, no_prompt)
 
-    data = deepcopy(data_manager.basic_config)
+    data = deepcopy(_data_manager.basic_config)
     name = _edit_instance_name(old_name, new_name, confirm_overwrite, no_prompt)
     _edit_data_path(data, name, data_path, copy_data, no_prompt)
 
@@ -237,7 +237,7 @@ def _edit_data_path(data, instance_name, data_path, copy_data, no_prompt):
         data["DATA_PATH"] = data_path
         if copy_data and not _copy_data(data):
             print("Can't copy data to non-empty location. Data location will remain unchanged.")
-            data["DATA_PATH"] = data_manager.basic_config["DATA_PATH"]
+            data["DATA_PATH"] = _data_manager.basic_config["DATA_PATH"]
     elif not no_prompt and confirm("Would you like to change the data location?", default=False):
         data["DATA_PATH"] = get_data_dir(
             instance_name=instance_name, data_path=None, interactive=True
@@ -246,7 +246,7 @@ def _edit_data_path(data, instance_name, data_path, copy_data, no_prompt):
             if not _copy_data(data):
                 print("Can't copy the data to non-empty location.")
                 if not confirm("Do you still want to use the new data location?"):
-                    data["DATA_PATH"] = data_manager.basic_config["DATA_PATH"]
+                    data["DATA_PATH"] = _data_manager.basic_config["DATA_PATH"]
                     print("Data location will remain unchanged.")
                     return
             print("Old data has been copied over to the new location.")
@@ -261,7 +261,7 @@ def _copy_data(data):
             # this is needed because copytree doesn't work when destination folder exists
             # Python 3.8 has `dirs_exist_ok` option for that
             os.rmdir(data["DATA_PATH"])
-    shutil.copytree(data_manager.basic_config["DATA_PATH"], data["DATA_PATH"])
+    shutil.copytree(_data_manager.basic_config["DATA_PATH"], data["DATA_PATH"])
     return True
 
 
@@ -279,7 +279,7 @@ def early_exit_runner(
             loop.run_until_complete(func())
             return
 
-        data_manager.load_basic_configuration(cli_flags.instance_name)
+        _data_manager.load_basic_configuration(cli_flags.instance_name)
         red = Red(cli_flags=cli_flags, description="Red V3", dm_help=None)
         driver_cls = _drivers.get_driver_class()
         loop.run_until_complete(driver_cls.initialize(**data_manager.storage_details()))
@@ -318,7 +318,7 @@ async def run_bot(red: Red, cli_flags: Namespace) -> None:
     )
 
     log.debug("====Basic Config====")
-    log.debug("Data Path: %s", data_manager._base_data_path())
+    log.debug("Data Path: %s", data_manager.data_path())
     log.debug("Storage Type: %s", data_manager.storage_type())
 
     # lib folder has to be in sys.path before trying to load any 3rd-party cog (GH-3061)
@@ -491,9 +491,9 @@ def main():
                 "\033[0m"
             )
             cli_flags.instance_name = "temporary_red"
-            data_manager.create_temp_config()
+            _data_manager.create_temp_config()
 
-        data_manager.load_basic_configuration(cli_flags.instance_name)
+        _data_manager.load_basic_configuration(cli_flags.instance_name)
 
         red = Red(cli_flags=cli_flags, description="Red V3", dm_help=None)
 

redbot/cogs/audio/core/tasks/lavalink.py

@@ -108,7 +108,7 @@ async def lavalink_attempt_connect(self, timeout: int = 50, manual: bool = False
                     password=password,
                     port=port,
                     timeout=timeout,
-                    resume_key=f"Red-Core-Audio-{self.bot.user.id}-{data_manager.instance_name}",
+                    resume_key=f"Red-Core-Audio-{self.bot.user.id}-{data_manager.instance_name()}",
                     secured=secured,
                 )
             except lavalink.AbortingNodeConnection:

redbot/core/_cli.py

@@ -7,8 +7,7 @@
 
 import discord
 from discord import __version__ as discord_version
-
-from redbot.core.utils._internal_utils import cli_level_to_log_level
+from red_commons.logging import VERBOSE, TRACE
 
 
 # This needs to be an int enum to be used
@@ -324,3 +323,15 @@ def parse_cli_flags(args):
     args.logging_level = cli_level_to_log_level(args.logging_level)
 
     return args
+
+
+def cli_level_to_log_level(level: int) -> int:
+    if level == 0:
+        log_level = logging.INFO
+    elif level == 1:
+        log_level = logging.DEBUG
+    elif level == 2:
+        log_level = VERBOSE
+    else:
+        log_level = TRACE
+    return log_level

redbot/core/_data_manager.py

@@ -0,0 +1,121 @@
+import json
+import os
+import sys
+import tempfile
+from copy import deepcopy
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+import platformdirs
+
+from ._cli import ExitCodes
+
+__all__ = (
+    "basic_config",
+    "instance_name",
+    "load_existing_config",
+    "create_temp_config",
+    "load_basic_configuration",
+)
+
+basic_config: Optional[Dict[str, Any]] = None
+
+instance_name = None
+
+basic_config_default: Dict[str, Any] = {
+    "DATA_PATH": None,
+    "COG_PATH_APPEND": "cogs",
+    "CORE_PATH_APPEND": "core",
+}
+
+appdir = platformdirs.PlatformDirs("Red-DiscordBot")
+config_dir = appdir.user_config_path
+_system_user = sys.platform == "linux" and 0 < os.getuid() < 1000
+if _system_user:
+    if Path.home().exists():
+        # We don't want to break someone just because they created home dir
+        # but were already using the site_data_path.
+        #
+        # But otherwise, we do want Red to use user_config_path if home dir exists.
+        _maybe_config_file = appdir.site_data_path / "config.json"
+        if _maybe_config_file.exists():
+            config_dir = _maybe_config_file.parent
+    else:
+        config_dir = appdir.site_data_path
+
+config_file = config_dir / "config.json"
+
+
+def load_existing_config():
+    """Get the contents of the config file, or an empty dictionary if it does not exist.
+
+    Returns
+    -------
+    dict
+        The config data.
+    """
+    if not config_file.exists():
+        return {}
+
+    with config_file.open(encoding="utf-8") as fs:
+        return json.load(fs)
+
+
+def create_temp_config():
+    """
+    Creates a default instance for Red, so it can be ran
+    without creating an instance.
+
+    .. warning:: The data of this instance will be removed
+        on next system restart.
+    """
+    name = "temporary_red"
+
+    default_dirs = deepcopy(basic_config_default)
+    default_dirs["DATA_PATH"] = tempfile.mkdtemp()
+    default_dirs["STORAGE_TYPE"] = "JSON"
+    default_dirs["STORAGE_DETAILS"] = {}
+
+    config = load_existing_config()
+
+    config[name] = default_dirs
+
+    with config_file.open("w", encoding="utf-8") as fs:
+        json.dump(config, fs, indent=4)
+
+
+def load_basic_configuration(instance_name_: str):
+    """Loads the basic bootstrap configuration necessary for `Config`
+    to know where to store or look for data.
+
+    .. important::
+        It is necessary to call this function BEFORE getting any `Config`
+        objects!
+
+    Parameters
+    ----------
+    instance_name_ : str
+        The instance name given by CLI argument and created during
+        redbot setup.
+    """
+    global basic_config
+    global instance_name
+    instance_name = instance_name_
+
+    try:
+        with config_file.open(encoding="utf-8") as fs:
+            config = json.load(fs)
+    except FileNotFoundError:
+        print(
+            "You need to configure the bot instance using `redbot-setup`"
+            " prior to running the bot."
+        )
+        sys.exit(ExitCodes.CONFIGURATION_ERROR)
+    try:
+        basic_config = config[instance_name]
+    except KeyError:
+        print(
+            "Instance with this name doesn't exist."
+            " You can create new instance using `redbot-setup` prior to running the bot."
+        )
+        sys.exit(ExitCodes.INVALID_CLI_USAGE)

redbot/core/_debuginfo.py

@@ -131,10 +131,10 @@ def _get_os_variables_section(self) -> DebugInfoSection:
         )
 
     async def _get_red_vars_section(self) -> DebugInfoSection:
-        if data_manager.instance_name is None:
+        if data_manager.instance_name() is None:
             return DebugInfoSection(
                 "Red variables",
-                f"Metadata file: {data_manager.config_file}",
+                f"Metadata file: {data_manager.metadata_file()}",
             )
 
         parts = [f"Instance name: {data_manager.instance_name}"]
@@ -161,8 +161,8 @@ async def _get_red_vars_section(self) -> DebugInfoSection:
             parts.append(f"Disabled intents: {disabled_intents}")
 
         parts.append(f"Storage type: {data_manager.storage_type()}")
-        parts.append(f"Data path: {data_manager.basic_config['DATA_PATH']}")
-        parts.append(f"Metadata file: {data_manager.config_file}")
+        parts.append(f"Data path: {data_manager.data_path()}")
+        parts.append(f"Metadata file: {data_manager.metadata_file()}")
 
         return DebugInfoSection(
             "Red variables",

redbot/core/core_commands.py

@@ -4675,10 +4675,9 @@ async def dm(self, ctx: commands.Context, user_id: int, *, message: str):
     @commands.is_owner()
     async def datapath(self, ctx: commands.Context):
         """Prints the bot's data path."""
-        from redbot.core.data_manager import basic_config
+        from redbot.core.data_manager import data_path
 
-        data_dir = Path(basic_config["DATA_PATH"])
-        msg = _("Data path: {path}").format(path=data_dir)
+        msg = _("Data path: {path}").format(path=data_path())
         await ctx.send(box(msg))
 
     @commands.command(hidden=True)