-
Notifications
You must be signed in to change notification settings - Fork 6
Expand file tree
/
Copy pathutils.py
More file actions
84 lines (60 loc) · 2.33 KB
/
utils.py
File metadata and controls
84 lines (60 loc) · 2.33 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
import os
import json
import logging
import re
from stackify import compat
internal_logger = logging.getLogger(__name__)
def arg_or_env(name, args, default=None, env_key=None):
env_name = env_key or 'STACKIFY_{0}'.format(name.upper())
try:
value = args.get(name)
if not value:
value = os.environ[env_name]
return value
except KeyError:
if default is not None:
return default
else:
raise NameError('You must specify the keyword argument {0} or environment variable {1}'.format(name, env_name))
def data_to_json(data):
try:
return json.dumps(data, default=lambda x: get_default_object(x))
except Exception as e:
internal_logger.exception('Failed to serialize object to json: {} - Exception: {}'.format(str(data), str(e)))
return str(data) # String representation of the object
def extract_request(data):
if 'request' in data and "WSGIRequest" in str(data['request']):
new_request = {}
obj = data['request']
if hasattr(obj, 'path'):
new_request['path'] = obj.path
if hasattr(obj, 'method'):
new_request['method'] = obj.method
if hasattr(obj, 'POST'):
new_request['form'] = obj.POST
if hasattr(obj, 'GET'):
new_request['query'] = obj.GET
if hasattr(obj, 'content_type'):
new_request['content_type'] = obj.content_type
data['request'] = new_request
return data
def get_default_object(obj):
if object_is_iterable(obj):
return [item for item in obj]
return hasattr(obj, '__dict__') and obj.__dict__ or obj.__str__()
def object_is_iterable(obj):
return hasattr(obj, '__iter__') or isinstance(obj, str)
class RegexValidator(object):
def __init__(self, regex, verbose_pattern=None):
self.regex = regex
self.verbose_pattern = verbose_pattern or regex
def __call__(self, value, field_name):
value = compat.text_type(value)
match = re.match(self.regex, value)
if match:
return value
raise ConfigError("{} does not match pattern {}".format(value, self.verbose_pattern), field_name)
class ConfigError(ValueError):
def __init__(self, msg, field_name):
self.field_name = field_name
super(ValueError, self).__init__(msg)