Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_reraise_requests_exception(self):
@urlmatch(netloc=r'(.*\.)?api\.weixin\.qq\.com$')
def _wechat_api_mock(url, request):
return {'status_code': 404, 'content': '404 not found'}
try:
with HTTMock(_wechat_api_mock):
self.client.material.get_count()
except WeChatClientException as e:
self.assertEqual(404, e.response.status_code)
def test_reraise_requests_exception(self):
@urlmatch(netloc=r'(.*\.)?api\.weixin\.qq\.com$')
def _wechat_api_mock(url, request):
return {'status_code': 404, 'content': '404 not found'}
try:
with HTTMock(_wechat_api_mock):
self.oauth.fetch_access_token('123456')
except WeChatClientException as e:
self.assertEqual(404, e.response.status_code)
method=method,
url_or_endpoint=url,
result_processor=result_processor,
**kwargs
)
elif errcode == WeChatErrorCode.OUT_OF_API_FREQ_LIMIT.value:
# api freq out of limit
raise APILimitedException(
errcode,
errmsg,
client=self,
request=res.request,
response=res
)
else:
raise WeChatClientException(
errcode,
errmsg,
client=self,
request=res.request,
response=res
)
return result if not result_processor else result_processor(result)
if 'errcode' in result and result['errcode'] != 0:
errcode = result['errcode']
errmsg = result['errmsg']
if errcode == 42001:
# access_token expired, fetch a new one and retry request
self.fetch_access_token()
return self._request(
method=method,
url=url,
**kwargs
)
elif errcode == 45009:
# api freq out of limit
raise APILimitedException(errcode, errmsg)
else:
raise WeChatClientException(errcode, errmsg)
return result
if isinstance(kwargs.get('data', ''), dict):
body = json.dumps(kwargs['data'], ensure_ascii=False)
body = body.encode('utf-8')
kwargs['data'] = body
kwargs['timeout'] = kwargs.get('timeout', self.timeout)
result_processor = kwargs.pop('result_processor', None)
res = self._http.request(
method=method,
url=url,
**kwargs
)
try:
res.raise_for_status()
except requests.RequestException as reqe:
raise WeChatClientException(
errcode=None,
errmsg=None,
client=self,
request=reqe.request,
response=reqe.response
)
return self._handle_result(
res, method, url, result_processor, **kwargs
)
def _fetch_access_token(self, url, params):
""" The real fetch access token """
logger.info('Fetching access token')
res = self._http.get(
url=url,
params=params
)
try:
res.raise_for_status()
except requests.RequestException as reqe:
raise WeChatClientException(
errcode=None,
errmsg=None,
client=self,
request=reqe.request,
response=reqe.response
)
result = res.json()
if 'errcode' in result and result['errcode'] != 0:
raise WeChatClientException(
result['errcode'],
result['errmsg'],
client=self,
request=res.request,
response=res
)
def __init__(self, errcode=-40005, errmsg='Invalid AppId'):
super(InvalidAppIdException, self).__init__(errcode, errmsg)
class WeChatOAuthException(WeChatClientException):
"""WeChat OAuth API exception class"""
pass
class WeChatComponentOAuthException(WeChatClientException):
"""WeChat Component OAuth API exception class"""
pass
class WeChatPayException(WeChatClientException):
"""WeChat Pay API exception class"""
def __init__(self, return_code, result_code=None, return_msg=None,
errcode=None, errmsg=None, client=None,
request=None, response=None):
"""
:param return_code: 返回状态码
:param result_code: 业务结果
:param return_msg: 返回信息
:param errcode: 错误代码
:param errmsg: 错误代码描述
"""
super(WeChatPayException, self).__init__(
errcode,
errmsg,
client,
def get_menu(self):
"""
查询自定义菜单。
详情请参考 http://mp.weixin.qq.com/wiki/index.php?title=自定义菜单查询接口
:return: 返回的 JSON 数据包
"""
try:
return self._get('menu/get')
except WeChatClientException as e:
if e.errcode == 46003:
# menu not exist
return None
else:
raise e
method=method,
url=url,
**kwargs
)
else:
res_future = aiohttp.request(
method=method,
url=url,
**kwargs
)
res = yield from asyncio.wait_for(res_future, timeout)
# reset kwargs for later retrying
kwargs['timeout'] = timeout
kwargs['files'] = files
except ClientError:
raise WeChatClientException(
errcode=None,
errmsg=None,
client=self,
)
# dirty hack
res = yield from self._decode_result(res)
return self._handle_result(
res, method, url, result_processor, **kwargs
)