# coding=utf-8 import logging from alipay.aop.api.AlipayClientConfig import AlipayClientConfig from alipay.aop.api.DefaultAlipayClient import DefaultAlipayClient from typing import TYPE_CHECKING from library.alipay import AliException if TYPE_CHECKING: from apps.web.core.models import AliApp logger = logging.getLogger(__name__) class AliApiProxy(object): URL = "https://openapi.alipay.com/gateway.do" URL_DEBUG = "https://openapi.alipaydev.com/gateway.do" def __init__(self, provider, debug=False): # type:(AliApp, bool) -> None self._url = self.URL if not debug else self.URL_DEBUG self._appid = str(provider.appid) self._app_private_key = provider.app_private_key_string self._alipay_public_key = provider.public_key_string self._client = None def __str__(self): return "AliApiProxy <{}_{}>".format(self.__class__.__name__, self._appid) @property def client(self): if not self._client: _config = AlipayClientConfig() _config.server_url = self._url _config.app_id = self._appid _config.app_private_key, _config.alipay_public_key = self._app_private_key, self._alipay_public_key self._client = DefaultAlipayClient(_config, logger=logger) return self._client def request(self, request, response): try: responseContent = self.client.execute(request) except Exception as e: logger.error(e.message) raise AliException(errCode="", errMsg=e.message) response.parse_response_content(responseContent) if not response.is_success(): raise AliException(errCode=response.code, errMsg=response.msg) return response