mirror of
https://github.com/nonebot/nonebot2.git
synced 2025-09-06 12:06:45 +00:00
⚡ improve plugin system (#1011)
This commit is contained in:
@ -35,14 +35,77 @@ FrontMatter:
|
||||
description: nonebot.plugin 模块
|
||||
"""
|
||||
|
||||
from typing import List, Optional
|
||||
from itertools import chain
|
||||
from types import ModuleType
|
||||
from contextvars import ContextVar
|
||||
from typing import Set, Dict, List, Optional
|
||||
|
||||
_plugins: Dict[str, "Plugin"] = {}
|
||||
_managers: List["PluginManager"] = []
|
||||
_current_plugin: ContextVar[Optional["Plugin"]] = ContextVar(
|
||||
"_current_plugin", default=None
|
||||
)
|
||||
|
||||
|
||||
def _module_name_to_plugin_name(module_name: str) -> str:
|
||||
return module_name.rsplit(".", 1)[-1]
|
||||
|
||||
|
||||
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
|
||||
return plugin
|
||||
|
||||
|
||||
def _revert_plugin(plugin: "Plugin") -> None:
|
||||
if plugin.name not in _plugins:
|
||||
raise RuntimeError("Plugin not found!")
|
||||
del _plugins[plugin.name]
|
||||
|
||||
|
||||
def get_plugin(name: str) -> Optional["Plugin"]:
|
||||
"""获取已经导入的某个插件。
|
||||
|
||||
如果为 `load_plugins` 文件夹导入的插件,则为文件(夹)名。
|
||||
|
||||
参数:
|
||||
name: 插件名,即 {ref}`nonebot.plugin.plugin.Plugin.name`。
|
||||
"""
|
||||
return _plugins.get(name)
|
||||
|
||||
|
||||
def get_plugin_by_module_name(module_name: str) -> Optional["Plugin"]:
|
||||
"""通过模块名获取已经导入的某个插件。
|
||||
|
||||
如果提供的模块名为某个插件的子模块,同样会返回该插件。
|
||||
|
||||
参数:
|
||||
module_name: 模块名,即 {ref}`nonebot.plugin.plugin.Plugin.module_name`。
|
||||
"""
|
||||
splits = module_name.split(".")
|
||||
loaded = {plugin.module_name: plugin for plugin in _plugins.values()}
|
||||
while splits:
|
||||
name = ".".join(splits)
|
||||
if name in loaded:
|
||||
return loaded[name]
|
||||
splits.pop()
|
||||
|
||||
|
||||
def get_loaded_plugins() -> Set["Plugin"]:
|
||||
"""获取当前已导入的所有插件。"""
|
||||
return set(_plugins.values())
|
||||
|
||||
|
||||
def get_available_plugin_names() -> Set[str]:
|
||||
"""获取当前所有可用的插件名(包含尚未加载的插件)。"""
|
||||
return {*chain.from_iterable(manager.available_plugins for manager in _managers)}
|
||||
|
||||
|
||||
from .on import on as on
|
||||
from .manager import PluginManager
|
||||
from .export import Export as Export
|
||||
@ -61,7 +124,6 @@ from .on import CommandGroup as CommandGroup
|
||||
from .on import MatcherGroup as MatcherGroup
|
||||
from .on import on_fullmatch as on_fullmatch
|
||||
from .on import on_metaevent as on_metaevent
|
||||
from .plugin import get_plugin as get_plugin
|
||||
from .load import load_plugins as load_plugins
|
||||
from .on import on_startswith as on_startswith
|
||||
from .load import load_from_json as load_from_json
|
||||
@ -69,5 +131,4 @@ from .load import load_from_toml as load_from_toml
|
||||
from .on import on_shell_command as on_shell_command
|
||||
from .load import load_all_plugins as load_all_plugins
|
||||
from .load import load_builtin_plugin as load_builtin_plugin
|
||||
from .plugin import get_loaded_plugins as get_loaded_plugins
|
||||
from .load import load_builtin_plugins as load_builtin_plugins
|
||||
|
Reference in New Issue
Block a user