94 lines
2.7 KiB
Python
94 lines
2.7 KiB
Python
"""模块：支付网关注册表
|
||
网关注册工厂——管理所有支付渠道适配器的注册与发现。
|
||
子服务通过注册表获取网关实例，不需要知道具体实现类。"""
|
||
|
||
from typing import Dict, Type, Optional
|
||
from gvsdsdk.payment.gateways.base import (
|
||
AbstractPaymentGateway,
|
||
PaymentCredentials,
|
||
PaymentChannel,
|
||
)
|
||
|
||
|
||
class GatewayRegistry:
|
||
"""支付网关注册工厂
|
||
|
||
用法:
|
||
registry = GatewayRegistry()
|
||
|
||
# 注册渠道
|
||
registry.register('wechat', WechatPaymentGateway)
|
||
|
||
# 创建网关实例
|
||
gw = registry.create('wechat', credentials)
|
||
resp = gw.Pay(request)
|
||
"""
|
||
|
||
def __init__(self):
|
||
self._gateways: Dict[str, Type[AbstractPaymentGateway]] = {}
|
||
|
||
def register(self, channel: str, gateway_class: Type[AbstractPaymentGateway]):
|
||
"""注册支付渠道适配器
|
||
|
||
Args:
|
||
channel: 渠道标识（如 'wechat', 'alipay', 'bank'）
|
||
gateway_class: 实现 AbstractPaymentGateway 的适配器类
|
||
"""
|
||
if not issubclass(gateway_class, AbstractPaymentGateway):
|
||
raise TypeError(
|
||
f'{gateway_class.__name__} 必须继承 AbstractPaymentGateway'
|
||
)
|
||
self._gateways[channel] = gateway_class
|
||
|
||
def unregister(self, channel: str):
|
||
"""注销支付渠道"""
|
||
self._gateways.pop(channel, None)
|
||
|
||
def create(self, channel: str, credentials: PaymentCredentials) -> AbstractPaymentGateway:
|
||
"""根据渠道标识创建网关实例
|
||
|
||
Args:
|
||
channel: 渠道标识
|
||
credentials: 支付凭证
|
||
|
||
Returns:
|
||
AbstractPaymentGateway 实例
|
||
|
||
Raises:
|
||
ValueError: 渠道未注册
|
||
"""
|
||
gateway_class = self._gateways.get(channel)
|
||
if gateway_class is None:
|
||
raise ValueError(
|
||
f'未注册的支付渠道: {channel}，可用渠道: {list(self._gateways.keys())}'
|
||
)
|
||
return gateway_class(credentials)
|
||
|
||
def list_channels(self):
|
||
"""列出所有已注册的支付渠道"""
|
||
return list(self._gateways.keys())
|
||
|
||
def get_gateway_info(self, channel: str):
|
||
"""获取渠道信息"""
|
||
gateway_class = self._gateways.get(channel)
|
||
if gateway_class is None:
|
||
return None
|
||
dummy = gateway_class
|
||
return {
|
||
'channel': dummy.Channel,
|
||
'name': dummy.Name,
|
||
}
|
||
|
||
def list_all_info(self):
|
||
"""列出所有渠道信息"""
|
||
return [
|
||
{
|
||
'channel': cls.Channel,
|
||
'name': cls.Name,
|
||
}
|
||
for cls in self._gateways.values()
|
||
]
|
||
|
||
|
||
payment_registry = GatewayRegistry()
|