Split public and private config/driver APIs (#6024)

.github/labeler.yml

@@ -162,6 +162,7 @@
   - any:
       - redbot/core/_drivers/**/*
       - "!redbot/core/_drivers/**/locales/*"
+  - redbot/core/_config.py
   - redbot/core/config.py
   # Docs
   - docs/framework_config.rst

redbot/core/_config.py

@@ -0,0 +1,26 @@
+import weakref
+from typing import Tuple, Type
+
+from redbot.core.config import Config, _config_cache
+from redbot.core._drivers import BaseDriver
+
+__all__ = ("get_latest_confs", "migrate")
+
+_retrieved = weakref.WeakSet()
+
+
+def get_latest_confs() -> Tuple[Config, ...]:
+    global _retrieved
+    ret = set(_config_cache.values()) - set(_retrieved)
+    _retrieved |= ret
+    return tuple(ret)
+
+
+async def migrate(cur_driver_cls: Type[BaseDriver], new_driver_cls: Type[BaseDriver]) -> None:
+    """Migrate from one driver type to another."""
+    # Get custom group data
+    core_conf = Config.get_core_conf(allow_old=True)
+    core_conf.init_custom("CUSTOM_GROUPS", 2)
+    all_custom_group_data = await core_conf.custom("CUSTOM_GROUPS").all()
+
+    await cur_driver_cls.migrate_to(new_driver_cls, all_custom_group_data)

redbot/core/_drivers/__init__.py

@@ -8,6 +8,8 @@ from .postgres import PostgresDriver
 
 __all__ = [
     "get_driver",
+    "get_driver_class",
+    "get_driver_class_include_old",
     "ConfigCategory",
     "IdentifierData",
     "BaseDriver",
@@ -32,7 +34,7 @@ class BackendType(enum.Enum):
 _DRIVER_CLASSES = {BackendType.JSON: JsonDriver, BackendType.POSTGRES: PostgresDriver}
 
 
-def _get_driver_class_include_old(storage_type: Optional[BackendType] = None) -> Type[BaseDriver]:
+def get_driver_class_include_old(storage_type: Optional[BackendType] = None) -> Type[BaseDriver]:
     """
     ONLY for use in CLI for moving data away from a no longer supported backend
     """
@@ -115,7 +117,7 @@ def get_driver(
         if not allow_old:
             driver_cls: Type[BaseDriver] = get_driver_class(storage_type)
         else:
-            driver_cls: Type[BaseDriver] = _get_driver_class_include_old(storage_type)
+            driver_cls: Type[BaseDriver] = get_driver_class_include_old(storage_type)
     except ValueError:
         if storage_type in (BackendType.MONGOV1, BackendType.MONGO):
             raise RuntimeError(

redbot/core/_events.py

@@ -21,7 +21,7 @@ from redbot.core.i18n import (
 )
 from .. import __version__ as red_version, version_info as red_version_info
 from . import commands
-from .config import get_latest_confs
+from ._config import get_latest_confs
 from .utils._internal_utils import (
     fuzzy_command_search,
     format_fuzzy_results,

redbot/core/config.py

@@ -35,7 +35,6 @@ log = logging.getLogger("red.config")
 _T = TypeVar("_T")
 
 _config_cache = weakref.WeakValueDictionary()
-_retrieved = weakref.WeakSet()
 
 
 class ConfigMeta(type):
@@ -65,14 +64,6 @@ class ConfigMeta(type):
         return instance
 
 
-def get_latest_confs() -> Tuple["Config"]:
-    global _retrieved
-    ret = set(_config_cache.values()) - set(_retrieved)
-    _retrieved |= ret
-    # noinspection PyTypeChecker
-    return tuple(ret)
-
-
 class _ValueCtxManager(Awaitable[_T], AsyncContextManager[_T]):  # pylint: disable=duplicate-bases
     """Context manager implementation of config values.
 
@@ -1529,16 +1520,6 @@ class Config(metaclass=ConfigMeta):
             return self._lock_cache.setdefault(id_data, asyncio.Lock())
 
 
-async def migrate(cur_driver_cls: Type[BaseDriver], new_driver_cls: Type[BaseDriver]) -> None:
-    """Migrate from one driver type to another."""
-    # Get custom group data
-    core_conf = Config.get_core_conf(allow_old=True)
-    core_conf.init_custom("CUSTOM_GROUPS", 2)
-    all_custom_group_data = await core_conf.custom("CUSTOM_GROUPS").all()
-
-    await cur_driver_cls.migrate_to(new_driver_cls, all_custom_group_data)
-
-
 def _str_key_dict(value: Dict[Any, _T]) -> Dict[str, _T]:
     """
     Recursively casts all keys in the given `dict` to `str`.

redbot/setup.py

@@ -20,10 +20,16 @@ from redbot.core.utils._internal_utils import (
     create_backup as red_create_backup,
     cli_level_to_log_level,
 )
-from redbot.core import config, data_manager, _drivers
+from redbot.core import config, data_manager
+from redbot.core._config import migrate
 from redbot.core._cli import ExitCodes
 from redbot.core.data_manager import appdir, config_dir, config_file
-from redbot.core._drivers import BackendType, IdentifierData
+from redbot.core._drivers import (
+    BackendType,
+    IdentifierData,
+    get_driver_class,
+    get_driver_class_include_old,
+)
 
 conversion_log = logging.getLogger("red.converter")
 
@@ -211,7 +217,7 @@ def basic_setup(
     storage_type = get_storage_type(backend, interactive=interactive)
 
     default_dirs["STORAGE_TYPE"] = storage_type.value
-    driver_cls = _drivers.get_driver_class(storage_type)
+    driver_cls = get_driver_class(storage_type)
     default_dirs["STORAGE_DETAILS"] = driver_cls.get_config_details()
 
     if name in instance_data:
@@ -262,15 +268,15 @@ def get_target_backend(backend: str) -> BackendType:
 async def do_migration(
     current_backend: BackendType, target_backend: BackendType
 ) -> Dict[str, Any]:
-    cur_driver_cls = _drivers._get_driver_class_include_old(current_backend)
-    new_driver_cls = _drivers.get_driver_class(target_backend)
+    cur_driver_cls = get_driver_class_include_old(current_backend)
+    new_driver_cls = get_driver_class(target_backend)
     cur_storage_details = data_manager.storage_details()
     new_storage_details = new_driver_cls.get_config_details()
 
     await cur_driver_cls.initialize(**cur_storage_details)
     await new_driver_cls.initialize(**new_storage_details)
 
-    await config.migrate(cur_driver_cls, new_driver_cls)
+    await migrate(cur_driver_cls, new_driver_cls)
 
     await cur_driver_cls.teardown()
     await new_driver_cls.teardown()
@@ -284,7 +290,7 @@ async def create_backup(instance: str, destination_folder: Path = Path.home()) -
     if backend_type != BackendType.JSON:
         await do_migration(backend_type, BackendType.JSON)
     print("Backing up the instance's data...")
-    driver_cls = _drivers.get_driver_class()
+    driver_cls = get_driver_class()
     await driver_cls.initialize(**data_manager.storage_details())
     backup_fpath = await red_create_backup(destination_folder)
     await driver_cls.teardown()
@@ -320,7 +326,7 @@ async def remove_instance(
     if _create_backup is True:
         await create_backup(instance)
 
-    driver_cls = _drivers.get_driver_class(backend)
+    driver_cls = get_driver_class(backend)
     if delete_data is True:
         await driver_cls.initialize(**data_manager.storage_details())
         try: