Feature: 嵌套插件名称作用域优化 (#2665)

This commit is contained in:
Ju4tCode
2024-04-20 14:47:12 +08:00
committed by GitHub
parent e15d544341
commit 6bf10aafb7
16 changed files with 262 additions and 131 deletions

View File

@ -50,8 +50,8 @@ C = TypeVar("C", bound=BaseModel)
_plugins: dict[str, "Plugin"] = {}
_managers: list["PluginManager"] = []
_current_plugin_chain: ContextVar[tuple["Plugin", ...]] = ContextVar(
"_current_plugin_chain", default=()
_current_plugin: ContextVar[Optional["Plugin"]] = ContextVar(
"_current_plugin", default=None
)
@ -59,34 +59,87 @@ def _module_name_to_plugin_name(module_name: str) -> str:
return module_name.rsplit(".", 1)[-1]
def _controlled_modules() -> dict[str, str]:
return {
plugin_id: module_name
for manager in _managers
for plugin_id, module_name in manager.controlled_modules.items()
}
def _find_parent_plugin_id(
module_name: str, controlled_modules: Optional[dict[str, str]] = None
) -> Optional[str]:
if controlled_modules is None:
controlled_modules = _controlled_modules()
available = {
module_name: plugin_id for plugin_id, module_name in controlled_modules.items()
}
while "." in module_name:
module_name, _ = module_name.rsplit(".", 1)
if module_name in available:
return available[module_name]
def _module_name_to_plugin_id(
module_name: str, controlled_modules: Optional[dict[str, str]] = None
) -> str:
plugin_name = _module_name_to_plugin_name(module_name)
if parent_plugin_id := _find_parent_plugin_id(module_name, controlled_modules):
return f"{parent_plugin_id}:{plugin_name}"
return plugin_name
def _new_plugin(
module_name: str, module: ModuleType, manager: "PluginManager"
) -> "Plugin":
plugin_name = _module_name_to_plugin_name(module_name)
if plugin_name in _plugins:
raise RuntimeError("Plugin already exists! Check your plugin name.")
plugin = Plugin(plugin_name, module, module_name, manager)
_plugins[plugin_name] = plugin
plugin_id = _module_name_to_plugin_id(module_name)
if plugin_id in _plugins:
raise RuntimeError(
f"Plugin {plugin_id} already exists! Check your plugin name."
)
parent_plugin_id = _find_parent_plugin_id(module_name)
if parent_plugin_id is not None and parent_plugin_id not in _plugins:
raise RuntimeError(
f"Parent plugin {parent_plugin_id} must "
f"be loaded before loading {plugin_id}."
)
parent_plugin = _plugins[parent_plugin_id] if parent_plugin_id is not None else None
plugin = Plugin(
name=_module_name_to_plugin_name(module_name),
module=module,
module_name=module_name,
manager=manager,
parent_plugin=parent_plugin,
)
if parent_plugin:
parent_plugin.sub_plugins.add(plugin)
_plugins[plugin_id] = plugin
return plugin
def _revert_plugin(plugin: "Plugin") -> None:
if plugin.name not in _plugins:
if plugin.id_ not in _plugins:
raise RuntimeError("Plugin not found!")
del _plugins[plugin.name]
del _plugins[plugin.id_]
if parent_plugin := plugin.parent_plugin:
parent_plugin.sub_plugins.remove(plugin)
parent_plugin.sub_plugins.discard(plugin)
def get_plugin(name: str) -> Optional["Plugin"]:
def get_plugin(plugin_id: str) -> Optional["Plugin"]:
"""获取已经导入的某个插件。
如果为 `load_plugins` 文件夹导入的插件,则为文件(夹)名。
如果为嵌套的子插件,标识符为 `父插件标识符:子插件文件(夹)名`。
参数:
name: 插件,即 {ref}`nonebot.plugin.model.Plugin.name`。
plugin_id: 插件标识符,即 {ref}`nonebot.plugin.model.Plugin.id_`。
"""
return _plugins.get(name)
return _plugins.get(plugin_id)
def get_plugin_by_module_name(module_name: str) -> Optional["Plugin"]:
@ -111,7 +164,7 @@ def get_loaded_plugins() -> set["Plugin"]:
def get_available_plugin_names() -> set[str]:
"""获取当前所有可用的插件(包含尚未加载的插件)。"""
"""获取当前所有可用的插件标识符(包含尚未加载的插件)。"""
return {*chain.from_iterable(manager.available_plugins for manager in _managers)}