102 lines
3.1 KiB
Python
102 lines
3.1 KiB
Python
"""云厂商 SDK 适配层 — 抽象基类与标准化数据结构
|
||
|
||
各云厂商适配器继承 BaseAdapter，实现统一接口，将平台特定的 API 返回
|
||
转换为标准化的资产结构（NormalizedVPS / NormalizedDomain / AccountInfo），
|
||
供同步服务写入数据库。这样上层逻辑无需关心各平台 API 差异。
|
||
"""
|
||
|
||
from abc import ABC, abstractmethod
|
||
from dataclasses import asdict, dataclass, field
|
||
from typing import Optional
|
||
|
||
|
||
@dataclass
|
||
class NormalizedVPS:
|
||
"""标准化的 VPS 实例"""
|
||
|
||
external_id: str
|
||
name: str
|
||
ip_address: Optional[str] = None
|
||
region: Optional[str] = None
|
||
os: Optional[str] = None
|
||
cpu_cores: Optional[int] = None
|
||
memory_gb: Optional[float] = None
|
||
disk_gb: Optional[int] = None
|
||
status: str = "active"
|
||
monthly_cost: Optional[float] = None
|
||
currency: str = "USD"
|
||
raw: dict = field(default_factory=dict)
|
||
|
||
def to_dict(self) -> dict:
|
||
return asdict(self)
|
||
|
||
|
||
@dataclass
|
||
class NormalizedDomain:
|
||
"""标准化的域名"""
|
||
|
||
external_id: str
|
||
domain_name: str
|
||
registrar: Optional[str] = None
|
||
expiry_date: Optional[str] = None # ISO 日期字符串 YYYY-MM-DD
|
||
status: str = "active"
|
||
raw: dict = field(default_factory=dict)
|
||
|
||
def to_dict(self) -> dict:
|
||
return asdict(self)
|
||
|
||
|
||
@dataclass
|
||
class AccountInfo:
|
||
"""标准化的账户信息"""
|
||
|
||
balance: Optional[float] = None
|
||
currency: str = "USD"
|
||
pending_charges: Optional[float] = None
|
||
raw: dict = field(default_factory=dict)
|
||
|
||
def to_dict(self) -> dict:
|
||
return asdict(self)
|
||
|
||
|
||
def _overridden(instance: "BaseAdapter", method_name: str) -> bool:
|
||
"""判断子类是否重写了某方法（用于能力探测）"""
|
||
return type(instance).__dict__.get(method_name) is not None
|
||
|
||
|
||
class BaseAdapter(ABC):
|
||
"""云厂商适配器抽象基类
|
||
|
||
子类需：
|
||
- 通过 @register("xxx-sdk") 注册到注册表
|
||
- 设置 required_config（所需凭证字段，供前端表单提示）
|
||
- 实现 test_connection；按需重写 list_vps / list_domains / get_account
|
||
"""
|
||
|
||
#: 适配器所需的配置字段名（如 ["api_key"] 或 ["access_key_id", "access_key_secret"]）
|
||
required_config: list = []
|
||
|
||
def __init__(self, config: Optional[dict] = None):
|
||
self.config = config or {}
|
||
|
||
@abstractmethod
|
||
def test_connection(self) -> dict:
|
||
"""测试连接 / 凭证有效性，返回 {"ok": bool, "message": str, ...}"""
|
||
|
||
def list_vps(self) -> list:
|
||
raise NotImplementedError(f"{type(self).__name__} 未实现 list_vps")
|
||
|
||
def list_domains(self) -> list:
|
||
raise NotImplementedError(f"{type(self).__name__} 未实现 list_domains")
|
||
|
||
def get_account(self) -> AccountInfo:
|
||
raise NotImplementedError(f"{type(self).__name__} 未实现 get_account")
|
||
|
||
def capabilities(self) -> dict:
|
||
"""返回适配器实际支持的能力（是否重写了相应方法）"""
|
||
return {
|
||
"list_vps": _overridden(self, "list_vps"),
|
||
"list_domains": _overridden(self, "list_domains"),
|
||
"get_account": _overridden(self, "get_account"),
|
||
}
|