(
    entrypoint_group: str,
    enabled_plugin: Optional[str] = None,
    config: Optional[configparser.ConfigParser] = None,
    clear_cache: bool = False,
)
return plugins
Load all storage plugins that are registered with pkg_resources

    Parameters
    ==========
    entrypoint_group: str
        The entrypoint group name to load plugins from
    enabled_plugin: str
        The optional enabled storage plugin to search for
    config: configparser.ConfigParser
        The optional configparser instance to pass in
    clear_cache: bool
        Whether to clear the plugin cache

    Returns
    =======
    List of Storage:
        A list of objects derived from the Storage class
Load all storage plugins that are registered with pkg_resources
Load all storage plugins that are registered with pkg_resources
def load_storage_plugins(
    entrypoint_group: str,
    enabled_plugin: Optional[str] = None,
    config: Optional[configparser.ConfigParser] = None,
    clear_cache: bool = False,
) -> Set[Storage]:
    """
    Load all storage plugins that are registered with pkg_resources

    Parameters
    ==========
    entrypoint_group: str
        The entrypoint group name to load plugins from
    enabled_plugin: str
        The optional enabled storage plugin to search for
    config: configparser.ConfigParser
        The optional configparser instance to pass in
    clear_cache: bool
        Whether to clear the plugin cache

    Returns
    =======
    List of Storage:
        A list of objects derived from the Storage class
    """
    global loaded_storage_plugins
    if config is None:
        config = BandersnatchConfig().config
    if not enabled_plugin:
        try:
            enabled_plugin = config["mirror"]["storage-backend"]
            logger.info(f"Loading storage plugin: {enabled_plugin}")
        except KeyError:
            enabled_plugin = "filesystem"
            logger.info(
                "Failed to find configured storage backend, using default: "
                f"{enabled_plugin}"
            )
            pass

    if clear_cache:
        loaded_storage_plugins = defaultdict(list)

    # If the plugins for the entrypoint_group have been loaded return them
    cached_plugins = loaded_storage_plugins.get(entrypoint_group)
    if cached_plugins:
        return set(cached_plugins)

    plugins = set()
    for entry_point in pkg_resources.iter_entry_points(group=entrypoint_group):
        if entry_point.name == enabled_plugin + "_plugin":
            try:
                plugin_class = entry_point.load()
                plugin_instance = plugin_class(config=config)
                plugins.add(plugin_instance)
            except ModuleNotFoundError as me:
                logger.error(f"Unable to load entry point {entry_point}: {me}")

    loaded_storage_plugins[entrypoint_group] = list(plugins)

    return plugins
def load_storage_plugins ( entrypoint_group : str , enabled_plugin : Optional [ str ] = None , config : Optional [ configparser . ConfigParser ] = None , clear_cache : bool = False , ) -> Set [ Storage ] : global loaded_storage_plugins if config is None : config = BandersnatchConfig ( ) . config if not enabled_plugin : try : enabled_plugin = config [ "mirror" ] [ "storage-backend" ] logger . info ( f"Loading storage plugin: {enabled_plugin}" ) except KeyError : enabled_plugin = "filesystem" logger . info ( "Failed to find configured storage backend, using default: " f"{enabled_plugin}" ) pass if clear_cache : loaded_storage_plugins = defaultdict ( list ) # If the plugins for the entrypoint_group have been loaded return them cached_plugins = loaded_storage_plugins . get ( entrypoint_group ) if cached_plugins : return set ( cached_plugins ) plugins = set ( ) for entry_point in pkg_resources . iter_entry_points ( group = entrypoint_group ) : if entry_point . name == enabled_plugin + "_plugin" : try : plugin_class = entry_point . load ( ) plugin_instance = plugin_class ( config = config ) plugins . add ( plugin_instance ) except ModuleNotFoundError as me : logger . error ( f"Unable to load entry point {entry_point}: {me}" ) loaded_storage_plugins [ entrypoint_group ] = list ( plugins ) return plugins