ServicesMgr.process_data() - Code Metrics - Inspection of "oauth1 and oauth2 in main service class" - foxmask/django-th - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 574502...0b3f92 )

by Fox

created 2016-06-12 20:56 UTC

ServicesMgr.process_data() A

↳ Parent: ServicesMgr

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	1
c	1
b	0
f	0
dl	0
loc	11
rs	9.4285

# coding: utf-8 -*-
# Using OAuth1Session
from requests_oauthlib import OAuth1Session, OAuth2Session

# django stuff
from django.core.cache import caches
from django.core.urlresolvers import reverse

try:
    from django.apps import apps
    get_model = apps.get_model
except ImportError:
    from django.db.models.loading import get_model

# django_th stuff
from django_th.models import UserService, ServicesActivated
from django_th.publishing_limit import PublishingLimit
from django_th.html_entities import HtmlEntities


class ServicesMgr(object):
    """
        Main Service Class
    """
    name = ''
    title = ''
    body = ''
    data = {}

    class __ServicesMgr:

        def __init__(self, arg):
            self.val = arg

        def __str__(self):
            return repr(self) + self.val

    instance = None

    def __init__(self, arg, **kwargs):
        base = 'https://www.urltotheserviceapi.com'
        self.AUTH_URL = '{}/api/rest/v1/oauth/authorize/'.format(base)
        self.REQ_TOKEN = '{}/api/rest/v1/oauth/request_token/'.format(base)
        self.ACC_TOKEN = '{}/api/rest/v1/oauth/access_token/'.format(base)
        self.oauth = 'oauth1'
        self.service = ''
        if not ServicesMgr.instance:
            ServicesMgr.instance = ServicesMgr.__ServicesMgr(arg)
        else:
            ServicesMgr.instance.val = arg

    def __getattr__(self, name):
        return getattr(self.instance, name)

    def __str__(self):
        return self.name

    @staticmethod
    def _get_content(data, which_content):
        """
            get the content that could be hidden
            in the middle of "content" or "summary detail"
            from the data of the provider
        """
        content = ''
        if data.get(which_content):
            if type(data.get(which_content)) is list or\
               type(data.get(which_content)) is tuple or\
               type(data.get(which_content)) is dict:
                if 'value' in data.get(which_content)[0]:
                    content = data.get(which_content)[0].value
            else:
                if type(data.get(which_content)) is str:
                    content = data.get(which_content)
                else:
                    # if not str or list or tuple
                    # or dict it could be feedparser.FeedParserDict
                    # so get the item value
                    content = data.get(which_content)['value']
        return content

    @staticmethod
    def set_title(data):
        """
            handle the title from the data
            :param data: contains the data from the provider
            :type data: dict
            :rtype: string
        """
        title = (data.get('title') if data.get('title') else data.get('link'))

        return title

    def set_content(self, data):
        """
            handle the content from the data
            :param data: contains the data from the provider
            :type data: dict
            :rtype: string
        """
        content = self._get_content(data, 'content')

        if content == '':
            content = self._get_content(data, 'summary_detail')

        if content == '':
            if data.get('description'):
                content = data.get('description')

        return content

    def read_data(self, **kwargs):
        """
            get the data from the service

            :param kwargs: contain keyword args : trigger_id and model name
            :type kwargs: dict
            :rtype: model
        """
        model = get_model('django_th', kwargs['model_name'])

        return model.objects.get(trigger_id=kwargs['trigger_id'])

    def process_data(self, **kwargs):
        """
             get the data from the cache
            :param kwargs: contain keyword args : trigger_id at least
            :type kwargs: dict
        """
        cache = caches[kwargs.get('cache_stack')]
        cache_data = cache.get(kwargs.get('cache_stack') + '_' +
                               kwargs.get('trigger_id'))
        return PublishingLimit.get_data(kwargs.get('cache_stack'),
                                        cache_data, kwargs.get('trigger_id'))

    def save_data(self, trigger_id, data, **kwargs):
        """
            used to save data to the service
            but first of all
            make some work about the data to find
            and the data to convert
            :param trigger_id: trigger ID from which to save data
            :param data: the data to check to be used and save
            :type trigger_id: int
            :type data:  dict
            :return: the status of the save statement
            :rtype: boolean
        """
        title = self.set_title(data)
        title = HtmlEntities(title).html_entity_decode
        content = self.set_content(data)
        content = HtmlEntities(content).html_entity_decode
        if kwargs.get('output_format'):
            # pandoc to convert tools
            import pypandoc
            content = pypandoc.convert(content,
                                       kwargs.get('output_format'),
                                       format='html')
        return title, content

    def auth(self, request):
        """
            get the auth of the services
            :param request: contains the current session
            :type request: dict
            :type oauth: string of the oauth version used
            :rtype: dict
        """
        request_token = self.get_request_token(request)

        return request_token

    def callback_url(self, request):
        """
            the url to go back after the external service call
            :param request: contains the current session
            :type request: dict
            :rtype: string
        """
        service = self.service.split('Service')[1].lower()
        return_to = '{service}_callback'.format(service=service)

        return '%s://%s%s' % (request.scheme, request.get_host(),
                              reverse(return_to))

    def callback(self, request, **kwargs):
        """
            Called from the Service when the user accept to activate it
            the url to go back after the external service call
            :param request: contains the current session
            :param kwargs: keyword args
            :type request: dict
            :type kwargs: dict
            :rtype: string
        """
        if self.oauth == 'oauth1':
            token = self.callback_oauth1(request, **kwargs)
        else:
            token = self.callback_oauth2(request)

        service_name = ServicesActivated.objects.get(name=self.service)

        UserService.objects.filter(user=request.user,
                                   name=service_name
                                   ).update(token=token)
        back = self.service.split('Service')[1].lower()
        back_to = '{back_to}/callback.html'.format(back_to=back)
        return back_to

    def callback_oauth1(self, request, **kwargs):
        """
            Process for oAuth 1
            :param request: contains the current session
            :param kwargs: keyword args
            :type request: dict
            :type kwargs: dict
            :rtype: string
        """
        if kwargs['access_token'] == '':
            access_token = self.get_access_token(
                request.session['oauth_token'],
                request.session['oauth_token_secret'],
                request.GET.get('oauth_verifier', '')
            )
        else:
            access_token = kwargs['access_token']

        if type(access_token) == str:
            token = access_token
        else:
            token = '#TH#'.join((access_token.get('oauth_token'),
                                 access_token.get('oauth_token_secret')))
        return token

    def callback_oauth2(self, request):
        """
            Process for oAuth 2
            :param request: contains the current session
            :return:
        """
        callback_url = self.callback_url(request)
        oauth = OAuth2Session(client_id=self.consumer_key,
                              redirect_uri=callback_url,
                              scope=self.scope)
        request_token = oauth.fetch_token(self.ACC_TOKEN,
                                          code=request.GET.get('code', ''),
                                          authorization_response=callback_url,
                                          client_secret=self.consumer_secret)
        return request_token.get('access_token')

    def get_request_token(self, request):
        """
           request the token to the external service
        """
        if self.oauth == 'oauth1':
            oauth = OAuth1Session(self.consumer_key,
                                  client_secret=self.consumer_secret)
            request_token = oauth.fetch_request_token(self.REQ_TOKEN)
            # Save the request token information for later
            request.session['oauth_token'] = request_token['oauth_token']
            request.session['oauth_token_secret'] = request_token[
                'oauth_token_secret']
            return request_token
        else:
            callback_url = self.callback_url(request)
            oauth = OAuth2Session(client_id=self.consumer_key,
                                  redirect_uri=callback_url,
                                  scope=self.scope)
            authorization_url, state = oauth.authorization_url(self.AUTH_URL)
            return authorization_url

    def get_access_token(self, oauth_token, oauth_token_secret,
                         oauth_verifier):
        """
           get the access token
            the url to go back after the external service call
            :param oauth_token: oauth token
            :param oauth_token_secret: oauth secret token
            :param oauth_verifier: oauth verifier
            :type oauth_token: string
            :type oauth_token_secret: string
            :type oauth_verifier: string
            :rtype: dict
        """
        # Using OAuth1Session
        oauth = OAuth1Session(self.consumer_key,
                              client_secret=self.consumer_secret,
                              resource_owner_key=oauth_token,
                              resource_owner_secret=oauth_token_secret,
                              verifier=oauth_verifier)
        oauth_tokens = oauth.fetch_access_token(self.ACC_TOKEN)

        return oauth_tokens


1			# coding: utf-8 -*-
2			# Using OAuth1Session
3			from requests_oauthlib import OAuth1Session, OAuth2Session
4
5			# django stuff
6			from django.core.cache import caches
7			from django.core.urlresolvers import reverse
8
9			try:
10			from django.apps import apps
11			get_model = apps.get_model
12			except ImportError:
13			from django.db.models.loading import get_model
14
15			# django_th stuff
16			from django_th.models import UserService, ServicesActivated
17			from django_th.publishing_limit import PublishingLimit
18			from django_th.html_entities import HtmlEntities
19
20
21			class ServicesMgr(object):
22			"""
23			Main Service Class
24			"""
25			name = ''
26			title = ''
27			body = ''
28			data = {}
29
30			class __ServicesMgr:
31
32			def __init__(self, arg):
33			self.val = arg
34
35			def __str__(self):
36			return repr(self) + self.val
37
38			instance = None
39
40			def __init__(self, arg, **kwargs):
41			base = 'https://www.urltotheserviceapi.com'
42			self.AUTH_URL = '{}/api/rest/v1/oauth/authorize/'.format(base)
43			self.REQ_TOKEN = '{}/api/rest/v1/oauth/request_token/'.format(base)
44			self.ACC_TOKEN = '{}/api/rest/v1/oauth/access_token/'.format(base)
45			self.oauth = 'oauth1'
46			self.service = ''
47			if not ServicesMgr.instance:
48			ServicesMgr.instance = ServicesMgr.__ServicesMgr(arg)
49			else:
50			ServicesMgr.instance.val = arg
51
52			def __getattr__(self, name):
53			return getattr(self.instance, name)
54
55			def __str__(self):
56			return self.name
57
58			@staticmethod
59			def _get_content(data, which_content):
60			"""
61			get the content that could be hidden
62			in the middle of "content" or "summary detail"
63			from the data of the provider
64			"""
65			content = ''
66			if data.get(which_content):
67			if type(data.get(which_content)) is list or\
68			type(data.get(which_content)) is tuple or\
69			type(data.get(which_content)) is dict:
70			if 'value' in data.get(which_content)[0]:
71			content = data.get(which_content)[0].value
72			else:
73			if type(data.get(which_content)) is str:
74			content = data.get(which_content)
75			else:
76			# if not str or list or tuple
77			# or dict it could be feedparser.FeedParserDict
78			# so get the item value
79			content = data.get(which_content)['value']
80			return content
81
82			@staticmethod
83			def set_title(data):
84			"""
85			handle the title from the data
86			:param data: contains the data from the provider
87			:type data: dict
88			:rtype: string
89			"""
90			title = (data.get('title') if data.get('title') else data.get('link'))
91
92			return title
93
94			def set_content(self, data):
95			"""
96			handle the content from the data
97			:param data: contains the data from the provider
98			:type data: dict
99			:rtype: string
100			"""
101			content = self._get_content(data, 'content')
102
103			if content == '':
104			content = self._get_content(data, 'summary_detail')
105
106			if content == '':
107			if data.get('description'):
108			content = data.get('description')
109
110			return content
111
112			def read_data(self, **kwargs):
113			"""
114			get the data from the service
115
116			:param kwargs: contain keyword args : trigger_id and model name
117			:type kwargs: dict
118			:rtype: model
119			"""
120			model = get_model('django_th', kwargs['model_name'])
121
122			return model.objects.get(trigger_id=kwargs['trigger_id'])
123
124			def process_data(self, **kwargs):
125			"""
126			get the data from the cache
127			:param kwargs: contain keyword args : trigger_id at least
128			:type kwargs: dict
129			"""
130			cache = caches[kwargs.get('cache_stack')]
131			cache_data = cache.get(kwargs.get('cache_stack') + '_' +
132			kwargs.get('trigger_id'))
133			return PublishingLimit.get_data(kwargs.get('cache_stack'),
134			cache_data, kwargs.get('trigger_id'))
135
136			def save_data(self, trigger_id, data, **kwargs):
137			"""
138			used to save data to the service
139			but first of all
140			make some work about the data to find
141			and the data to convert
142			:param trigger_id: trigger ID from which to save data
143			:param data: the data to check to be used and save
144			:type trigger_id: int
145			:type data: dict
146			:return: the status of the save statement
147			:rtype: boolean
148			"""
149			title = self.set_title(data)
150			title = HtmlEntities(title).html_entity_decode
151			content = self.set_content(data)
152			content = HtmlEntities(content).html_entity_decode
153			if kwargs.get('output_format'):
154			# pandoc to convert tools
155			import pypandoc
156			content = pypandoc.convert(content,
157			kwargs.get('output_format'),
158			format='html')
159			return title, content
160
161			def auth(self, request):
162			"""
163			get the auth of the services
164			:param request: contains the current session
165			:type request: dict
166			:type oauth: string of the oauth version used
167			:rtype: dict
168			"""
169			request_token = self.get_request_token(request)
170
171			return request_token
172
173			def callback_url(self, request):
174			"""
175			the url to go back after the external service call
176			:param request: contains the current session
177			:type request: dict
178			:rtype: string
179			"""
180			service = self.service.split('Service')[1].lower()
181			return_to = '{service}_callback'.format(service=service)
182
183			return '%s://%s%s' % (request.scheme, request.get_host(),
184			reverse(return_to))
185
186			def callback(self, request, **kwargs):
187			"""
188			Called from the Service when the user accept to activate it
189			the url to go back after the external service call
190			:param request: contains the current session
191			:param kwargs: keyword args
192			:type request: dict
193			:type kwargs: dict
194			:rtype: string
195			"""
196			if self.oauth == 'oauth1':
197			token = self.callback_oauth1(request, **kwargs)
198			else:
199			token = self.callback_oauth2(request)
200
201			service_name = ServicesActivated.objects.get(name=self.service)
202
203			UserService.objects.filter(user=request.user,
204			name=service_name
205			).update(token=token)
206			back = self.service.split('Service')[1].lower()
207			back_to = '{back_to}/callback.html'.format(back_to=back)
208			return back_to
209
210			def callback_oauth1(self, request, **kwargs):
211			"""
212			Process for oAuth 1
213			:param request: contains the current session
214			:param kwargs: keyword args
215			:type request: dict
216			:type kwargs: dict
217			:rtype: string
218			"""
219			if kwargs['access_token'] == '':
220			access_token = self.get_access_token(
221			request.session['oauth_token'],
222			request.session['oauth_token_secret'],
223			request.GET.get('oauth_verifier', '')
224			)
225			else:
226			access_token = kwargs['access_token']
227
228			if type(access_token) == str:
229			token = access_token
230			else:
231			token = '#TH#'.join((access_token.get('oauth_token'),
232			access_token.get('oauth_token_secret')))
233			return token
234
235			def callback_oauth2(self, request):
236			"""
237			Process for oAuth 2
238			:param request: contains the current session
239			:return:
240			"""
241			callback_url = self.callback_url(request)
242			oauth = OAuth2Session(client_id=self.consumer_key,
243			redirect_uri=callback_url,
244			scope=self.scope)
245			request_token = oauth.fetch_token(self.ACC_TOKEN,
246			code=request.GET.get('code', ''),
247			authorization_response=callback_url,
248			client_secret=self.consumer_secret)
249			return request_token.get('access_token')
250
251			def get_request_token(self, request):
252			"""
253			request the token to the external service
254			"""
255			if self.oauth == 'oauth1':
256			oauth = OAuth1Session(self.consumer_key,
257			client_secret=self.consumer_secret)
258			request_token = oauth.fetch_request_token(self.REQ_TOKEN)
259			# Save the request token information for later
260			request.session['oauth_token'] = request_token['oauth_token']
261			request.session['oauth_token_secret'] = request_token[
262			'oauth_token_secret']
263			return request_token
264			else:
265			callback_url = self.callback_url(request)
266			oauth = OAuth2Session(client_id=self.consumer_key,
267			redirect_uri=callback_url,
268			scope=self.scope)
269			authorization_url, state = oauth.authorization_url(self.AUTH_URL)
270			return authorization_url
271
272			def get_access_token(self, oauth_token, oauth_token_secret,
273			oauth_verifier):
274			"""
275			get the access token
276			the url to go back after the external service call
277			:param oauth_token: oauth token
278			:param oauth_token_secret: oauth secret token
279			:param oauth_verifier: oauth verifier
280			:type oauth_token: string
281			:type oauth_token_secret: string
282			:type oauth_verifier: string
283			:rtype: dict
284			"""
285			# Using OAuth1Session
286			oauth = OAuth1Session(self.consumer_key,
287			client_secret=self.consumer_secret,
288			resource_owner_key=oauth_token,
289			resource_owner_secret=oauth_token_secret,
290			verifier=oauth_verifier)
291			oauth_tokens = oauth.fetch_access_token(self.ACC_TOKEN)
292
293			return oauth_tokens
294

foxmask / django-th

Push — master ( 574502...0b3f92 )

ServicesMgr.process_data() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like