-
-
Notifications
You must be signed in to change notification settings - Fork 11
Expand file tree
/
Copy pathmixins.py
More file actions
executable file
·173 lines (143 loc) · 4.61 KB
/
mixins.py
File metadata and controls
executable file
·173 lines (143 loc) · 4.61 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
# -*- coding: utf-8 -*-
#
# Exception formatter mixin classes
#
# :copyright: 2020 Sonu Kumar
# :license: BSD-3-Clause
#
import abc
class MaskingMixin(object):
__metaclass__ = abc.ABCMeta
def __init__(self, mask_with, mask_key_has):
"""
:param mask_with: masked value to be used for a given variable
:param mask_key_has: tuple of strings to be used for checking masking rule
"""
self.mask_key_has = mask_key_has
self.mask_with = mask_with
@abc.abstractmethod
def __call__(self, key):
raise NotImplementedError
class ModelMixin(object):
"""
Base interface for data mode which can be used to store data in MongoDB/MySQL or any other data store.
"""
hash = None
host = None
path = None
method = None
request_data = None
exception_name = None
traceback = None
count = None
created_on = None
last_seen = None
def __str__(self):
return "'%s' '%s' %s" % (self.host, self.path, self.count)
def __unicode__(self):
return self.__str__()
def __repr__(self):
return "ModelMixin(%s)" % self.__str__()
@classmethod
@abc.abstractmethod
def delete_entity(cls, rhash):
"""
:param rhash: lookup key
:return: None
"""
raise NotImplementedError
@classmethod
@abc.abstractmethod
def create_or_update_entity(cls, rhash, host, path, method, request_data,
exception_name, traceback):
"""
:param rhash: Key of the db entry
:param host: App host e.g. example.com
:param path: request path
:param method: request method (GET/POST/PUT etc)
:param request_data: request form data
:param exception_name: exception name
:param traceback: Exception data captured
:return: error model object
"""
raise NotImplementedError
@classmethod
@abc.abstractmethod
def get_exceptions_per_page(cls, page_number=1):
"""
An object having these properties,
has_next, next_num, has_prev, prev_num and items
on the returned object like SQLAlchemy's Pagination
:return: a paginated object
"""
raise NotImplementedError
@classmethod
@abc.abstractmethod
def get_entity(cls, rhash):
"""
:param rhash: key for lookup
:return: Single entry of this class
"""
raise NotImplementedError
class NotificationMixin(object):
"""
A notification mixin class which can be used to notify any kind of notification.
"""
__metaclass__ = abc.ABCMeta
def __init__(self, *args, **kwargs):
pass
@abc.abstractmethod
def notify(self, request, exception,
email_subject=None,
email_body=None,
from_email=None,
recipient_list=None):
"""
:param request: current request context
:param exception : exception model object
:param email_subject: email subject
:param email_body: email message object
:param from_email: email sender as per config
:param recipient_list: list of email ids for notification
:return: None
"""
raise NotImplementedError
class ContextBuilderMixin(object):
"""
A context builder mixing that can be used to provide custom context details.
This will provide entire details, that would be used for DB logging
for usage see default context builder DefaultFlaskContextBuilder
"""
__metaclass__ = abc.ABCMeta
def __init__(self, *args, **kwargs):
pass
@abc.abstractmethod
def get_context(self, request, masking=None,
additional_context=None):
raise NotImplementedError
class TicketingMixin(object):
"""
A mixing class that would be called with model object to raise the ticket.
It can be used to directly create or update ticket in ticketing system
"""
__metaclass__ = abc.ABCMeta
def __init__(self, *args, **kwargs):
pass
@abc.abstractmethod
def raise_ticket(self, exception, request=None):
"""
:param exception: exception model object
:param request: current request
:return: None
"""
raise NotImplementedError
class ViewPermissionMixin(object):
__metaclass__ = abc.ABCMeta
@abc.abstractmethod
def __call__(self, request):
"""
Check whether user has the permission to view the request
:param request: request object
:return: True/False
"""
raise NotImplementedError