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

Completed

Push — master ( f067d1...61dc9d )

by Fox

created 2016-05-15 16:58 UTC

ServiceWallabag._new_wall() A

↳ Parent: ServiceWallabag

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	10
rs	9.4285

# 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

"""
    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 _new_wall(self, token):
        """
            new Wallabag instance
            :param token:
            :return: 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, userservice_id, url, title, tags, shootagain=False):
        """
            Create an entry
            :param userservice_id: id of the service to update with new token
            :param url:  url to save
            :param title: title to set
            :param tags: tags to set
            :param shootagain: retry to create a post after a 401
            :return: status
        """
        if shootagain:
            new_token = self.refresh_token()
            new_wall = self._new_wall(new_token)
            # update the token
            UserService.objects.filter(id=userservice_id).update(token=new_token)
        try:
            if shootagain:
                status = new_wall.post_entries(url=url,
                                               title=title,
                                               tags=tags
                                               )
            else:
                status = self.wall.post_entries(url=url,
                                                title=title,
                                                tags=tags)
            sentence = str('wallabag {} created').format(url)
            logger.debug(sentence)
        except Exception as e:
            if e.errno == 401:
                if shootagain:
                    # break circular call :P
                    logger.critical(e.errno, e.strerror)
                    status = False
                self._create_entry(userservice_id=userservice_id,
                                   url=url,
                                   title=title,
                                   tags=tags,
                                   shootagain=True)
            else:
                logger.critical(e.errno, e.strerror)
                status = False
        return status

    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['trigger_id']
        cache.set('th_wallabag_' + str(trigger_id), data)

    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
        """
        if 'link' in data and data['link'] is not None:
            if len(data['link']) > 0:
                # get the wallabag data for this trigger
                from th_wallabag.models import Wallabag
                trigger = Wallabag.objects.get(trigger_id=trigger_id)

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

                status = self._create_entry(userservice_id=data['userservice_id'],
                                            url=data['link'],
                                            title=title,
                                            tags=trigger.tag.lower())
            else:
                logger.warning(
                    "no link provided for trigger ID {}, so we ignore it".format(trigger_id))
                status = True
        else:
            logger.critical(
                "no token provided for trigger ID {}".format(trigger_id))
            status = False

        return status

    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'

    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)


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

foxmask / django-th

Push — master ( f067d1...61dc9d )

ServiceWallabag._new_wall() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like