wechatpy.client.base._is_api_endpoint() - Code Metrics - jxtech/wechatpy - Measure and Improve Code Quality continuously with Scrutinizer

wechatpy.client.base._is_api_endpoint() A
last analyzed 2020-03-06 10:31 UTC

↳ Parent: wechatpy.client.base

Complexity

Conditions

Size

Total Lines	2
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	1

Importance

Changes

Metric	Value
cc	1
eloc	2
nop	1
dl	0
loc	2
rs	10
c	0
b	0
f	0
ccs	2
cts	2
cp	1
crap	1

# -*- coding: utf-8 -*-
from __future__ import absolute_import, unicode_literals
import time
import inspect
import logging
import warnings

import six
import requests

from wechatpy.constants import WeChatErrorCode
from wechatpy.utils import json, get_querystring
from wechatpy.session.memorystorage import MemoryStorage
from wechatpy.exceptions import WeChatClientException, APILimitedException
from wechatpy.client.api.base import BaseWeChatAPI


logger = logging.getLogger(__name__)


def _is_api_endpoint(obj):
    return isinstance(obj, BaseWeChatAPI)


class BaseWeChatClient(object):
    API_BASE_URL = ''

    def __new__(cls, *args, **kwargs):
        self = super(BaseWeChatClient, cls).__new__(cls)
        api_endpoints = inspect.getmembers(self, _is_api_endpoint)
        for name, api in api_endpoints:
            api_cls = type(api)
            api = api_cls(self)
            setattr(self, name, api)
        return self

    def __init__(self, appid, access_token=None, session=None, timeout=None, auto_retry=True):
        self._http = requests.Session()
        self.appid = appid
        self.expires_at = None
        self.session = session or MemoryStorage()
        self.timeout = timeout
        self.auto_retry = auto_retry

        if isinstance(session, six.string_types):

            from shove import Shove
            from wechatpy.session.shovestorage import ShoveStorage

            querystring = get_querystring(session)
            prefix = querystring.get('prefix', ['wechatpy'])[0]

            shove = Shove(session)
            storage = ShoveStorage(shove, prefix)
            self.session = storage

        if access_token:
            self.session.set(self.access_token_key, access_token)

    @property
    def access_token_key(self):
        return '{0}_access_token'.format(self.appid)

    def _request(self, method, url_or_endpoint, **kwargs):
        if not url_or_endpoint.startswith(('http://', 'https://')):
            api_base_url = kwargs.pop('api_base_url', self.API_BASE_URL)
            url = '{base}{endpoint}'.format(
                base=api_base_url,
                endpoint=url_or_endpoint
            )
        else:
            url = url_or_endpoint

        if 'params' not in kwargs:
            kwargs['params'] = {}
        if isinstance(kwargs['params'], dict) and \
                'access_token' not in kwargs['params']:
            kwargs['params']['access_token'] = self.access_token
        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 _decode_result(self, res):
        try:
            result = json.loads(res.content.decode('utf-8', 'ignore'), strict=False)
        except (TypeError, ValueError):
            # Return origin response object if we can not decode it as JSON
            logger.debug('Can not decode response as JSON', exc_info=True)
            return res
        return result

    def _handle_result(self, res, method=None, url=None,
                       result_processor=None, **kwargs):
        if not isinstance(res, dict):
            # Dirty hack around asyncio based AsyncWeChatClient
            result = self._decode_result(res)
        else:
            result = res

        if not isinstance(result, dict):
            return result

        if 'base_resp' in result:
            # Different response in device APIs. Fuck tencent!
            result.update(result.pop('base_resp'))
        if 'errcode' in result:
            result['errcode'] = int(result['errcode'])

        if 'errcode' in result and result['errcode'] != 0:
            errcode = result['errcode']
            errmsg = result.get('errmsg', errcode)
            if self.auto_retry and errcode in (
                    WeChatErrorCode.INVALID_CREDENTIAL.value,
                    WeChatErrorCode.INVALID_ACCESS_TOKEN.value,
                    WeChatErrorCode.EXPIRED_ACCESS_TOKEN.value):
                logger.info('Access token expired, fetch a new one and retry request')
                self.fetch_access_token()
                access_token = self.session.get(self.access_token_key)
                kwargs['params']['access_token'] = access_token
                return self._request(
                    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)

    def get(self, url, **kwargs):
        return self._request(
            method='get',
            url_or_endpoint=url,
            **kwargs
        )

    def _get(self, url, **kwargs):
        warnings.warn('`_get` method of `WeChatClient` is deprecated, will be removed in 1.6,'
                      'Use `get` instead',
                      DeprecationWarning, stacklevel=2)
        return self.get(url, **kwargs)

    def post(self, url, **kwargs):
        return self._request(
            method='post',
            url_or_endpoint=url,
            **kwargs
        )

    def _post(self, url, **kwargs):
        warnings.warn('`_post` method of `WeChatClient` is deprecated, will be removed in 1.6,'
                      'Use `post` instead',
                      DeprecationWarning, stacklevel=2)
        return self.post(url, **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
            )

        expires_in = 7200
        if 'expires_in' in result:
            expires_in = result['expires_in']
        self.session.set(
            self.access_token_key,
            result['access_token'],
            expires_in
        )
        self.expires_at = int(time.time()) + expires_in
        return result

    def fetch_access_token(self):
        raise NotImplementedError()

    @property
    def access_token(self):
        """ WeChat access token """
        access_token = self.session.get(self.access_token_key)
        if access_token:
            if not self.expires_at:
                # user provided access_token, just return it
                return access_token

            timestamp = time.time()
            if self.expires_at - timestamp > 60:
                return access_token

        self.fetch_access_token()
        return self.session.get(self.access_token_key)


1			# -- coding: utf-8 --
2	10		from __future__ import absolute_import, unicode_literals
3	10		import time
4	10		import inspect
5	10		import logging
6	10		import warnings
7
8	10		import six
9	10		import requests
10
11	10		from wechatpy.constants import WeChatErrorCode
12	10		from wechatpy.utils import json, get_querystring
13	10		from wechatpy.session.memorystorage import MemoryStorage
14	10		from wechatpy.exceptions import WeChatClientException, APILimitedException
15	10		from wechatpy.client.api.base import BaseWeChatAPI
16
17
18	10		logger = logging.getLogger(__name__)
19
20
21	10		def _is_api_endpoint(obj):
22	10		return isinstance(obj, BaseWeChatAPI)
23
24
25	10		class BaseWeChatClient(object):
26	10		API_BASE_URL = ''
27
28	10		def __new__(cls, args, *kwargs):
29	10		self = super(BaseWeChatClient, cls).__new__(cls)
30	10		api_endpoints = inspect.getmembers(self, _is_api_endpoint)
31	10		for name, api in api_endpoints:
32	10		api_cls = type(api)
33	10		api = api_cls(self)
34	10		setattr(self, name, api)
35	10		return self
36
37	10		def __init__(self, appid, access_token=None, session=None, timeout=None, auto_retry=True):
38	10		self._http = requests.Session()
39	10		self.appid = appid
40	10		self.expires_at = None
41	10		self.session = session or MemoryStorage()
42	10		self.timeout = timeout
43	10		self.auto_retry = auto_retry
44
45	10	View Code Duplication	if isinstance(session, six.string_types):
			0 ignored issues – show Duplication introduced 2020-03-06 10:29 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
46	10		from shove import Shove
47	10		from wechatpy.session.shovestorage import ShoveStorage
48
49	10		querystring = get_querystring(session)
50	10		prefix = querystring.get('prefix', ['wechatpy'])[0]
51
52	10		shove = Shove(session)
53	10		storage = ShoveStorage(shove, prefix)
54	10		self.session = storage
55
56	10		if access_token:
57	10		self.session.set(self.access_token_key, access_token)
58
59	10		@property
60			def access_token_key(self):
61	10		return '{0}_access_token'.format(self.appid)
62
63	10		def _request(self, method, url_or_endpoint, **kwargs):
64	10		if not url_or_endpoint.startswith(('http://', 'https://')):
65	10		api_base_url = kwargs.pop('api_base_url', self.API_BASE_URL)
66	10		url = '{base}{endpoint}'.format(
67			base=api_base_url,
68			endpoint=url_or_endpoint
69			)
70			else:
71	10		url = url_or_endpoint
72
73	10		if 'params' not in kwargs:
74	10		kwargs['params'] = {}
75	10		if isinstance(kwargs['params'], dict) and \
76			'access_token' not in kwargs['params']:
77	10		kwargs['params']['access_token'] = self.access_token
78	10		if isinstance(kwargs.get('data', ''), dict):
79	10		body = json.dumps(kwargs['data'], ensure_ascii=False)
80	10		body = body.encode('utf-8')
81	10		kwargs['data'] = body
82
83	10		kwargs['timeout'] = kwargs.get('timeout', self.timeout)
84	10		result_processor = kwargs.pop('result_processor', None)
85	10		res = self._http.request(
86			method=method,
87			url=url,
88			**kwargs
89			)
90	10		try:
91	10		res.raise_for_status()
92			except requests.RequestException as reqe:
93			raise WeChatClientException(
94			errcode=None,
95			errmsg=None,
96			client=self,
97			request=reqe.request,
98			response=reqe.response
99			)
100
101	10		return self._handle_result(
102			res, method, url, result_processor, **kwargs
103			)
104
105	10		def _decode_result(self, res):
106	10		try:
107	10		result = json.loads(res.content.decode('utf-8', 'ignore'), strict=False)
108			except (TypeError, ValueError):
109			# Return origin response object if we can not decode it as JSON
110			logger.debug('Can not decode response as JSON', exc_info=True)
111			return res
112	10		return result
113
114	10		def _handle_result(self, res, method=None, url=None,
115			result_processor=None, **kwargs):
116	10		if not isinstance(res, dict):
117			# Dirty hack around asyncio based AsyncWeChatClient
118	10		result = self._decode_result(res)
119			else:
120			result = res
121
122	10		if not isinstance(result, dict):
123			return result
124
125	10		if 'base_resp' in result:
126			# Different response in device APIs. Fuck tencent!
127			result.update(result.pop('base_resp'))
128	10		if 'errcode' in result:
129	10		result['errcode'] = int(result['errcode'])
130
131	10		if 'errcode' in result and result['errcode'] != 0:
132			errcode = result['errcode']
133			errmsg = result.get('errmsg', errcode)
134			if self.auto_retry and errcode in (
135			WeChatErrorCode.INVALID_CREDENTIAL.value,
136			WeChatErrorCode.INVALID_ACCESS_TOKEN.value,
137			WeChatErrorCode.EXPIRED_ACCESS_TOKEN.value):
138			logger.info('Access token expired, fetch a new one and retry request')
139			self.fetch_access_token()
140			access_token = self.session.get(self.access_token_key)
141			kwargs['params']['access_token'] = access_token
142			return self._request(
143			method=method,
144			url_or_endpoint=url,
145			result_processor=result_processor,
146			**kwargs
147			)
148			elif errcode == WeChatErrorCode.OUT_OF_API_FREQ_LIMIT.value:
149			# api freq out of limit
150			raise APILimitedException(
151			errcode,
152			errmsg,
153			client=self,
154			request=res.request,
155			response=res
156			)
157			else:
158			raise WeChatClientException(
159			errcode,
160			errmsg,
161			client=self,
162			request=res.request,
163			response=res
164			)
165
166	10		return result if not result_processor else result_processor(result)
167
168	10		def get(self, url, **kwargs):
169	10		return self._request(
170			method='get',
171			url_or_endpoint=url,
172			**kwargs
173			)
174
175	10		def _get(self, url, **kwargs):
176			warnings.warn('`_get` method of `WeChatClient` is deprecated, will be removed in 1.6,'
177			'Use `get` instead',
178			DeprecationWarning, stacklevel=2)
179			return self.get(url, **kwargs)
180
181	10		def post(self, url, **kwargs):
182	10		return self._request(
183			method='post',
184			url_or_endpoint=url,
185			**kwargs
186			)
187
188	10		def _post(self, url, **kwargs):
189			warnings.warn('`_post` method of `WeChatClient` is deprecated, will be removed in 1.6,'
190			'Use `post` instead',
191			DeprecationWarning, stacklevel=2)
192			return self.post(url, **kwargs)
193
194	10	View Code Duplication	def _fetch_access_token(self, url, params):
			0 ignored issues – show Duplication introduced 2020-03-06 10:29 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
195			""" The real fetch access token """
196	10		logger.info('Fetching access token')
197	10		res = self._http.get(
198			url=url,
199			params=params
200			)
201	10		try:
202	10		res.raise_for_status()
203	10		except requests.RequestException as reqe:
204	10		raise WeChatClientException(
205			errcode=None,
206			errmsg=None,
207			client=self,
208			request=reqe.request,
209			response=reqe.response
210			)
211	10		result = res.json()
212	10		if 'errcode' in result and result['errcode'] != 0:
213			raise WeChatClientException(
214			result['errcode'],
215			result['errmsg'],
216			client=self,
217			request=res.request,
218			response=res
219			)
220
221	10		expires_in = 7200
222	10		if 'expires_in' in result:
223	10		expires_in = result['expires_in']
224	10		self.session.set(
225			self.access_token_key,
226			result['access_token'],
227			expires_in
228			)
229	10		self.expires_at = int(time.time()) + expires_in
230	10		return result
231
232	10		def fetch_access_token(self):
233			raise NotImplementedError()
234
235	10		@property
236			def access_token(self):
237			""" WeChat access token """
238	10		access_token = self.session.get(self.access_token_key)
239	10		if access_token:
240	10		if not self.expires_at:
241			# user provided access_token, just return it
242	10		return access_token
243
244	10		timestamp = time.time()
245	10		if self.expires_at - timestamp > 60:
246	10		return access_token
247
248	10		self.fetch_access_token()
249			return self.session.get(self.access_token_key)
250

jxtech / wechatpy

wechatpy.client.base._is_api_endpoint() A last analyzed 2020-03-06 10:31 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

wechatpy.client.base._is_api_endpoint() A
last analyzed 2020-03-06 10:31 UTC