"""适配器注册表与工厂 通过 @register("sdk-type") 装饰器将适配器类登记到注册表, 上层按 Provider.sdk_type 取得对应适配器实例。 """ from typing import Optional from app.adapters.base import BaseAdapter _REGISTRY: dict = {} def register(*sdk_types: str): """装饰器:将适配器类注册到一个或多个 sdk_type""" def decorator(cls): for t in sdk_types: _REGISTRY[t] = cls cls.sdk_types = list(sdk_types) return cls return decorator def load_all() -> None: """导入所有适配器模块以触发注册(幂等)""" from app.adapters import ( # noqa: F401 ai, aliyun, cloudflare, digitalocean, tencent, vultr, ) def get_adapter(sdk_type: str, config: Optional[dict] = None) -> BaseAdapter: """按 sdk_type 创建适配器实例""" load_all() cls = _REGISTRY.get(sdk_type) if not cls: raise ValueError(f"未支持的 SDK 类型:{sdk_type}") return cls(config) def is_supported(sdk_type: Optional[str]) -> bool: load_all() return sdk_type in _REGISTRY def supported_types() -> list: load_all() return sorted(_REGISTRY.keys()) def sdk_capabilities() -> dict: """返回 sdk_type -> capabilities 映射(供前端判断测试/同步按钮)""" load_all() result = {} for sdk_type, cls in _REGISTRY.items(): try: result[sdk_type] = cls({}).capabilities() except Exception: # noqa: BLE001 result[sdk_type] = {} return result def adapter_meta() -> list: """返回所有已注册适配器的元信息(供前端展示所需凭证字段与能力)""" load_all() seen = set() result = [] for sdk_type, cls in sorted(_REGISTRY.items()): if cls.__name__ in seen: continue seen.add(cls.__name__) result.append( { "class": cls.__name__, "sdk_types": getattr(cls, "sdk_types", [sdk_type]), "required_config": getattr(cls, "required_config", []), } ) return result