51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
|
"""Provider 自动发现 + 注册表"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import importlib
|
||
|
|
import pkgutil
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import TYPE_CHECKING
|
||
|
|
|
||
|
|
if TYPE_CHECKING:
|
||
|
|
from app.providers.base import BaseProvider, Capability
|
||
|
|
|
||
|
|
|
||
|
|
class ProviderRegistry:
|
||
|
|
"""全局 Provider 注册表，启动时自动扫描 providers/ 目录"""
|
||
|
|
|
||
|
|
_providers: dict[str, "BaseProvider"] = {}
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def auto_discover(cls):
|
||
|
|
"""扫描当前包下所有模块，注册 BaseProvider 子类实例"""
|
||
|
|
from app.providers.base import BaseProvider as _Base
|
||
|
|
|
||
|
|
package_dir = Path(__file__).parent
|
||
|
|
for info in pkgutil.iter_modules([str(package_dir)]):
|
||
|
|
if info.name in ("base", "__init__"):
|
||
|
|
continue
|
||
|
|
mod = importlib.import_module(f"app.providers.{info.name}")
|
||
|
|
for attr_name in dir(mod):
|
||
|
|
attr = getattr(mod, attr_name)
|
||
|
|
if (
|
||
|
|
isinstance(attr, type)
|
||
|
|
and issubclass(attr, _Base)
|
||
|
|
and attr is not _Base
|
||
|
|
and getattr(attr, "name", "")
|
||
|
|
):
|
||
|
|
instance = attr()
|
||
|
|
cls._providers[instance.name] = instance
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def get(cls, name: str) -> "BaseProvider | None":
|
||
|
|
return cls._providers.get(name)
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def all(cls) -> dict[str, "BaseProvider"]:
|
||
|
|
return dict(cls._providers)
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
def by_capability(cls, cap: "Capability") -> list["BaseProvider"]:
|
||
|
|
return [p for p in cls._providers.values() if cap in p.capabilities]
|