ServiceWallabag._new_token() - Code Metrics - Inspection of "some improvements" - foxmask/django-th - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 29b212...65a8c1 )

by Fox

created 2016-05-24 20:12 UTC

ServiceWallabag._new_token() A

↳ Parent: ServiceWallabag

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

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

# coding: utf-8
# add here the call of any native lib of python like datetime etc.

# add the python API here if needed
from wallabag_api.wallabag import Wallabag as Wall
# django classes
from django.core.urlresolvers import reverse
from django.utils.log import getLogger
from django.core.cache import caches

# django_th classes
from django_th.services.services import ServicesMgr
from django_th.html_entities import HtmlEntities
from django_th.models import UserService, ServicesActivated
from th_wallabag.models import Wallabag

"""
    handle process with wallabag
    put the following in settings.py

    TH_SERVICES = (
        ...
        'th_wallabag.my_wallabag.ServiceWallabag',
        ...
    )
"""

logger = getLogger('django_th.trigger_happy')

cache = caches['th_wallabag']


class ServiceWallabag(ServicesMgr):

    def __init__(self, token=None):
        super(ServiceWallabag, self).__init__(token)
        self.token = token
        if token:
            try:
                us = UserService.objects.get(token=token, name='ServiceWallabag')

                self.service_username = us.username
                self.service_password = us.password
                self.service_host = us.host
                self.service_client_secret = us.client_secret
                self.service_client_id = us.client_id

                self.wall = Wall(host=self.service_host,
                                 client_secret=self.service_client_secret,
                                 client_id=self.service_client_id,
                                 token=self.token)

            except UserService.DoesNotExist:
                pass

    def read_data(self, **kwargs):
        """
            get the data from the service
            as the pocket service does not have any date
            in its API linked to the note,
            add the triggered date to the dict data
            thus the service will be triggered when data will be found

            :param kwargs: contain keyword args : trigger_id at least
            :type kwargs: dict

            :rtype: list
        """
        data = list()
        trigger_id = kwargs.get('trigger_id')
        cache.set('th_wallabag_' + str(trigger_id), data)

    def new_wall(self, token):
        """
            produce a new wall instance
            :param token: to get
            :return: Wall instance
        """
        return Wall(host=self.service_host,
                    client_secret=self.service_client_secret,
                    client_id=self.service_client_id,
                    token=token)

    def _create_entry(self, title, data, tags):
        """
            create an entry
            :param title: string
            :param data: dict
            :param tags: list
            :return: boolean
        """
        if data.get('link') and len(data.get('link')) > 0:
            try:
                self.wall.post_entries(url=data.get('link'),
                                       title=title,
                                       tags=(tags.lower()))
                sentence = str('wallabag {} created').format(data.get('link'))
                logger.debug(sentence)
                status = True
            except AttributeError as e:
                    # in the __init__ the token was not found in the UserService model
                    status = self._new_token(data.get('userservice_id'),
                                             data.get('link'),
                                             title,
                                             tags.lower())
            except Exception as e:
                if e.errno == 401:
                    status = self._new_token(data.get('userservice_id'),
                                             data.get('link'),
                                             title,
                                             tags.lower())
                else:
                    logger.critical(e.errno, e.strerror)
                    status = False
        else:
            status = False
        return status

    def _refresh_token(self):
        """
            refresh the expired token
            :return: boolean
        """
        params = {'username': self.service_username,
                  'password': self.service_password,
                  'client_id': self.service_client_id,
                  'client_secret': self.service_client_secret}
        return Wall.get_token(host=self.service_host, **params)

    def _new_token(self, userservice_id, link, title, tags):
        """
            create a new token
            :param userservice_id: id of the UserService
            :param link: string
            :param title: string
            :param tags: list
            :return: boolean
        """
        new_token = self._refresh_token()
        UserService.objects.filter(id=userservice_id).update(token=new_token)
    
        # new wallabag session with new token
        new_wall = self.new_wall(new_token)
        try:
            return new_wall.post_entries(url=link, title=title, tags=tags)
        except Exception as e:
            logger.critical(e.errno, e.strerror)
            return False

    def save_data(self, trigger_id, **data):
        """
            let's save the data

            :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
        """
        trigger = Wallabag.objects.get(trigger_id=trigger_id)

        title = self.set_title(data)
        # convert htmlentities
        title = HtmlEntities(title).html_entity_decode

        return self._create_entry(title, data, trigger.tag)

    def auth(self, request):
        """
            let's auth the user to the Service
            :param request: request object
            :return: callback url
            :rtype: string that contains the url to redirect after auth
 
        """
        service = UserService.objects.get(user=request.user, name='ServiceWallabag')
        callback_url = 'http://%s%s' % (
            request.get_host(), reverse('wallabag_callback'))
        params = {'username': service.username,
                  'password': service.password,
                  'client_id': service.client_id,
                  'client_secret': service.client_secret}
        access_token = Wall.get_token(host=service.host, **params)
        request.session['oauth_token'] = access_token
        return callback_url

    def callback(self, request, **kwargs):
        """
            Called from the Service when the user accept to activate it
            :param request: request object
            :return: callback url
            :rtype: string , path to the template
        """

        try:
            UserService.objects.filter(
                user=request.user,
                name=ServicesActivated.objects.get(name='ServiceWallabag')
            )
        except KeyError:
            return '/'

        return 'wallabag/callback.html'


1			# coding: utf-8
2			# add here the call of any native lib of python like datetime etc.
3
4			# add the python API here if needed
5			from wallabag_api.wallabag import Wallabag as Wall
6			# django classes
7			from django.core.urlresolvers import reverse
8			from django.utils.log import getLogger
9			from django.core.cache import caches
10
11			# django_th classes
12			from django_th.services.services import ServicesMgr
13			from django_th.html_entities import HtmlEntities
14			from django_th.models import UserService, ServicesActivated
15			from th_wallabag.models import Wallabag
16
17			"""
18			handle process with wallabag
19			put the following in settings.py
20
21			TH_SERVICES = (
22			...
23			'th_wallabag.my_wallabag.ServiceWallabag',
24			...
25			)
26			"""
27
28			logger = getLogger('django_th.trigger_happy')
29
30			cache = caches['th_wallabag']
31
32
33			class ServiceWallabag(ServicesMgr):
34
35			def __init__(self, token=None):
36			super(ServiceWallabag, self).__init__(token)
37			self.token = token
38			if token:
39			try:
40			us = UserService.objects.get(token=token, name='ServiceWallabag')
41
42			self.service_username = us.username
43			self.service_password = us.password
44			self.service_host = us.host
45			self.service_client_secret = us.client_secret
46			self.service_client_id = us.client_id
47
48			self.wall = Wall(host=self.service_host,
49			client_secret=self.service_client_secret,
50			client_id=self.service_client_id,
51			token=self.token)
52
53			except UserService.DoesNotExist:
54			pass
55
56			def read_data(self, **kwargs):
57			"""
58			get the data from the service
59			as the pocket service does not have any date
60			in its API linked to the note,
61			add the triggered date to the dict data
62			thus the service will be triggered when data will be found
63
64			:param kwargs: contain keyword args : trigger_id at least
65			:type kwargs: dict
66
67			:rtype: list
68			"""
69			data = list()
70			trigger_id = kwargs.get('trigger_id')
71			cache.set('th_wallabag_' + str(trigger_id), data)
72
73			def new_wall(self, token):
74			"""
75			produce a new wall instance
76			:param token: to get
77			:return: Wall instance
78			"""
79			return Wall(host=self.service_host,
80			client_secret=self.service_client_secret,
81			client_id=self.service_client_id,
82			token=token)
83
84			def _create_entry(self, title, data, tags):
85			"""
86			create an entry
87			:param title: string
88			:param data: dict
89			:param tags: list
90			:return: boolean
91			"""
92			if data.get('link') and len(data.get('link')) > 0:
93			try:
94			self.wall.post_entries(url=data.get('link'),
95			title=title,
96			tags=(tags.lower()))
97			sentence = str('wallabag {} created').format(data.get('link'))
98			logger.debug(sentence)
99			status = True
100			except AttributeError as e:
101			# in the __init__ the token was not found in the UserService model
102			status = self._new_token(data.get('userservice_id'),
103			data.get('link'),
104			title,
105			tags.lower())
106			except Exception as e:
107			if e.errno == 401:
108			status = self._new_token(data.get('userservice_id'),
109			data.get('link'),
110			title,
111			tags.lower())
112			else:
113			logger.critical(e.errno, e.strerror)
114			status = False
115			else:
116			status = False
117			return status
118
119			def _refresh_token(self):
120			"""
121			refresh the expired token
122			:return: boolean
123			"""
124			params = {'username': self.service_username,
125			'password': self.service_password,
126			'client_id': self.service_client_id,
127			'client_secret': self.service_client_secret}
128			return Wall.get_token(host=self.service_host, **params)
129
130			def _new_token(self, userservice_id, link, title, tags):
131			"""
132			create a new token
133			:param userservice_id: id of the UserService
134			:param link: string
135			:param title: string
136			:param tags: list
137			:return: boolean
138			"""
139			new_token = self._refresh_token()
140			UserService.objects.filter(id=userservice_id).update(token=new_token)
141
142			# new wallabag session with new token
143			new_wall = self.new_wall(new_token)
144			try:
145			return new_wall.post_entries(url=link, title=title, tags=tags)
146			except Exception as e:
147			logger.critical(e.errno, e.strerror)
148			return False
149
150			def save_data(self, trigger_id, **data):
151			"""
152			let's save the data
153
154			:param trigger_id: trigger ID from which to save data
155			:param data: the data to check to be used and save
156			:type trigger_id: int
157			:type data: dict
158			:return: the status of the save statement
159			:rtype: boolean
160			"""
161			trigger = Wallabag.objects.get(trigger_id=trigger_id)
162
163			title = self.set_title(data)
164			# convert htmlentities
165			title = HtmlEntities(title).html_entity_decode
166
167			return self._create_entry(title, data, trigger.tag)
168
169			def auth(self, request):
170			"""
171			let's auth the user to the Service
172			:param request: request object
173			:return: callback url
174			:rtype: string that contains the url to redirect after auth
175
176			"""
177			service = UserService.objects.get(user=request.user, name='ServiceWallabag')
178			callback_url = 'http://%s%s' % (
179			request.get_host(), reverse('wallabag_callback'))
180			params = {'username': service.username,
181			'password': service.password,
182			'client_id': service.client_id,
183			'client_secret': service.client_secret}
184			access_token = Wall.get_token(host=service.host, **params)
185			request.session['oauth_token'] = access_token
186			return callback_url
187
188			def callback(self, request, **kwargs):
189			"""
190			Called from the Service when the user accept to activate it
191			:param request: request object
192			:return: callback url
193			:rtype: string , path to the template
194			"""
195
196			try:
197			UserService.objects.filter(
198			user=request.user,
199			name=ServicesActivated.objects.get(name='ServiceWallabag')
200			)
201			except KeyError:
202			return '/'
203
204			return 'wallabag/callback.html'
205

foxmask / django-th

Push — master ( 29b212...65a8c1 )

ServiceWallabag._new_token() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like