data_filter() - Code Metrics - Inspection of "Trigger Happy and Taiga.io Kan ban fixed #155" - foxmask/django-th - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 69fa9b...92a354 )

by Fox

created 2016-12-06 22:44 UTC

data_filter() F

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

Metric	Value
cc	52
c	1
b	0
f	0
dl	0
loc	65
rs	3.0769

How to fix Long Method Complexity

import hmac
import hashlib

from rest_framework.response import Response
from rest_framework.decorators import api_view

from django_th.models import TriggerService
from django_th.services import default_provider

from th_taiga.models import Taiga


def data_filter(trigger_id, **data):
    """
    check if we want to track event for a given action
    :param trigger_id:
    :param data:
    :return:
    """
    taiga = Taiga.objects.get(trigger_id=trigger_id)

    action = data.get('action')
    type = data.get('type')
    data = data.get('data')

    if type == 'epic' and action == 'create' and taiga.notify_epic_create:
        data['type_action'] = 'New Epic created'
    elif type == 'epic' and action == 'change' and taiga.notify_epic_change:
        data['type_action'] = 'Changed Epic'
    elif type == 'epic' and action == 'delete' and taiga.notify_epic_delete:
        data['type_action'] = 'Deleted Epic'
    elif type == 'issue' and action == 'create' and \
            taiga.notify_issue_create:
        data['type_action'] = 'New Issue created'
    elif type == 'issue' and action == 'change' and \
            taiga.notify_issue_change:
        data['type_action'] = 'Changed Issue'
    elif type == 'issue' and action == 'delete' and \
            taiga.notify_issue_delete:
        data['type_action'] = 'Deleted Issue'
    elif type == 'userstory' and action == 'create' and \
            taiga.notify_userstory_create:
        data['type_action'] = 'New Userstory created'
    elif type == 'userstory' and action == 'change' and \
            taiga.notify_userstory_change:
        data['type_action'] = 'Changed Userstory'
    elif type == 'userstory' and action == 'delete' and \
            taiga.notify_userstory_delete:
        data['type_action'] = 'Deleted Userstory'
    elif type == 'task' and action == 'create' and \
            taiga.notify_task_create:
        data['type_action'] = 'New Task created'
    elif type == 'task' and action == 'change' and \
            taiga.notify_task_change:
        data['type_action'] = 'Changed Task'
    elif type == 'task' and action == 'delete' and \
            taiga.notify_task_delete:
        data['type_action'] = 'Deleted Task'
    elif type == 'wikipage' and action == 'create' and \
            taiga.notify_wikipage_create:
        data['type_action'] = 'New Wikipage created'
    elif type == 'wikipage' and action == 'change' and \
            taiga.notify_wikipage_change:
        data['type_action'] = 'Changed Wikipage'
    elif type == 'wikipage' and action == 'delete' and \
            taiga.notify_wikipage_delete:
        data['type_action'] = 'Deleted Wikipage'
    elif type == 'relateduserstory' and action == 'create' and \
            taiga.notify_relateduserstory_create:
        data['type_action'] = 'New Related Userstory created'
    elif type == 'relateduserstory' and action == 'delete' and \
            taiga.notify_relateduserstory_delete:
        data['type_action'] = 'Deleted Related Userstory'
    else:
        data = []

    return data


def consumer(trigger_id, data):

    """
        call the consumer and handle the data
        :param trigger_id:
        :param data:
        :return:
    """
    # consumer - the service which uses the data
    default_provider.load_services()
    service = TriggerService.objects.get(id=trigger_id)

    service_consumer = default_provider.get_service(
        str(service.consumer.name.name))
    kwargs = {'user': service.user}

    data = data_filter(trigger_id, **data)
    if len(data) > 0:

        getattr(service_consumer, '__init__')(service.consumer.token,
                                              **kwargs)
        getattr(service_consumer, 'save_data')(service.id, **data)


def verify_signature(data, key, signature):
    mac = hmac.new(key.encode("utf-8"), msg=data, digestmod=hashlib.sha1)
    return mac.hexdigest() == signature


@api_view(['POST'])
def taiga(request, trigger_id, key):
    signature = request.META.get('HTTP_X_TAIGA_WEBHOOK_SIGNATURE')
    # check that the data are ok with the provided signature
    if verify_signature(request._request.body, key, signature):
        consumer(trigger_id, request.data)
        return Response({"message": "Success"})
    else:
        return Response({"message": "Failed!"})


1		import hmac
2		import hashlib
3
4		from rest_framework.response import Response
5		from rest_framework.decorators import api_view
6
7		from django_th.models import TriggerService
8		from django_th.services import default_provider
9
10		from th_taiga.models import Taiga
11
12
13		def data_filter(trigger_id, **data):
14		"""
15		check if we want to track event for a given action
16		:param trigger_id:
17		:param data:
18		:return:
19		"""
20		taiga = Taiga.objects.get(trigger_id=trigger_id)
21
22		action = data.get('action')
23		type = data.get('type')
24		data = data.get('data')
25
26		if type == 'epic' and action == 'create' and taiga.notify_epic_create:
27		data['type_action'] = 'New Epic created'
28		elif type == 'epic' and action == 'change' and taiga.notify_epic_change:
29		data['type_action'] = 'Changed Epic'
30		elif type == 'epic' and action == 'delete' and taiga.notify_epic_delete:
31		data['type_action'] = 'Deleted Epic'
32		elif type == 'issue' and action == 'create' and \
33		taiga.notify_issue_create:
34		data['type_action'] = 'New Issue created'
35		elif type == 'issue' and action == 'change' and \
36		taiga.notify_issue_change:
37		data['type_action'] = 'Changed Issue'
38		elif type == 'issue' and action == 'delete' and \
39		taiga.notify_issue_delete:
40		data['type_action'] = 'Deleted Issue'
41		elif type == 'userstory' and action == 'create' and \
42		taiga.notify_userstory_create:
43		data['type_action'] = 'New Userstory created'
44		elif type == 'userstory' and action == 'change' and \
45		taiga.notify_userstory_change:
46		data['type_action'] = 'Changed Userstory'
47		elif type == 'userstory' and action == 'delete' and \
48		taiga.notify_userstory_delete:
49		data['type_action'] = 'Deleted Userstory'
50		elif type == 'task' and action == 'create' and \
51		taiga.notify_task_create:
52		data['type_action'] = 'New Task created'
53		elif type == 'task' and action == 'change' and \
54		taiga.notify_task_change:
55		data['type_action'] = 'Changed Task'
56		elif type == 'task' and action == 'delete' and \
57		taiga.notify_task_delete:
58		data['type_action'] = 'Deleted Task'
59		elif type == 'wikipage' and action == 'create' and \
60		taiga.notify_wikipage_create:
61		data['type_action'] = 'New Wikipage created'
62		elif type == 'wikipage' and action == 'change' and \
63		taiga.notify_wikipage_change:
64		data['type_action'] = 'Changed Wikipage'
65		elif type == 'wikipage' and action == 'delete' and \
66		taiga.notify_wikipage_delete:
67		data['type_action'] = 'Deleted Wikipage'
68		elif type == 'relateduserstory' and action == 'create' and \
69		taiga.notify_relateduserstory_create:
70		data['type_action'] = 'New Related Userstory created'
71		elif type == 'relateduserstory' and action == 'delete' and \
72		taiga.notify_relateduserstory_delete:
73		data['type_action'] = 'Deleted Related Userstory'
74		else:
75		data = []
76
77		return data
78
79
80	View Code Duplication	def consumer(trigger_id, data):
		0 ignored issues – show Duplication introduced 2016-12-06 22:45 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Loading history...
81		"""
82		call the consumer and handle the data
83		:param trigger_id:
84		:param data:
85		:return:
86		"""
87		# consumer - the service which uses the data
88		default_provider.load_services()
89		service = TriggerService.objects.get(id=trigger_id)
90
91		service_consumer = default_provider.get_service(
92		str(service.consumer.name.name))
93		kwargs = {'user': service.user}
94
95		data = data_filter(trigger_id, **data)
96		if len(data) > 0:
97
98		getattr(service_consumer, '__init__')(service.consumer.token,
99		**kwargs)
100		getattr(service_consumer, 'save_data')(service.id, **data)
101
102
103		def verify_signature(data, key, signature):
104		mac = hmac.new(key.encode("utf-8"), msg=data, digestmod=hashlib.sha1)
105		return mac.hexdigest() == signature
106
107
108		@api_view(['POST'])
109		def taiga(request, trigger_id, key):
110		signature = request.META.get('HTTP_X_TAIGA_WEBHOOK_SIGNATURE')
111		# check that the data are ok with the provided signature
112		if verify_signature(request._request.body, key, signature):
113		consumer(trigger_id, request.data)
114		return Response({"message": "Success"})
115		else:
116		return Response({"message": "Failed!"})
117

foxmask / django-th

Push — master ( 69fa9b...92a354 )

data_filter() F

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like