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
|
import base64
import re
from io import BytesIO, StringIO
from jinja2 import filters
from json import dumps
from flask import redirect
from werkzeug.routing import HTTPException, RoutingException
import six
from datetime import datetime, timedelta
import csv
def slugify(value):
"""Normalizes string, converts to lowercase, removes non-alpha characters,
and converts spaces to hyphens.
Copy/Pasted from ametaireau/pelican/utils itself took from django sources.
"""
if isinstance(value, six.text_type):
import unicodedata
value = unicodedata.normalize('NFKD', value)
if six.PY2:
value = value.encode('ascii', 'ignore')
value = six.text_type(re.sub('[^\w\s-]', '', value).strip().lower())
return re.sub('[-\s]+', '-', value)
class Redirect303(HTTPException, RoutingException):
"""Raise if the map requests a redirect. This is for example the case if
`strict_slashes` are activated and an url that requires a trailing slash.
The attribute `new_url` contains the absolute destination url.
"""
code = 303
def __init__(self, new_url):
RoutingException.__init__(self, new_url)
self.new_url = new_url
def get_response(self, environ):
return redirect(self.new_url, 303)
class PrefixedWSGI(object):
'''
Wrap the application in this middleware and configure the
front-end server to add these headers, to let you quietly bind
this to a URL other than / and to an HTTP scheme that is
different than what is used locally.
It relies on "APPLICATION_ROOT" app setting.
Inspired from http://flask.pocoo.org/snippets/35/
:param app: the WSGI application
'''
def __init__(self, app):
self.app = app
self.wsgi_app = app.wsgi_app
def __call__(self, environ, start_response):
script_name = self.app.config['APPLICATION_ROOT']
if script_name:
environ['SCRIPT_NAME'] = script_name
path_info = environ['PATH_INFO']
if path_info.startswith(script_name):
environ['PATH_INFO'] = path_info[len(script_name):]
scheme = environ.get('HTTP_X_SCHEME', '')
if scheme:
environ['wsgi.url_scheme'] = scheme
return self.wsgi_app(environ, start_response)
def minimal_round(*args, **kw):
""" Jinja2 filter: rounds, but display only non-zero decimals
from http://stackoverflow.com/questions/28458524/
"""
# Use the original round filter, to deal with the extra arguments
res = filters.do_round(*args, **kw)
# Test if the result is equivalent to an integer and
# return depending on it
ires = int(res)
return (res if res != ires else ires)
def list_of_dicts2json(dict_to_convert):
"""Take a list of dictionnaries and turns it into
a json in-memory file
"""
return BytesIO(dumps(dict_to_convert).encode('utf-8'))
def list_of_dicts2csv(dict_to_convert):
"""Take a list of dictionnaries and turns it into
a csv in-memory file, assume all dict have the same keys
"""
# CSV writer has a different behavior in PY2 and PY3
# http://stackoverflow.com/a/37974772
try:
if six.PY3:
csv_file = StringIO()
# using list() for py3.4 compat. Otherwise, writerows() fails
# (expecting a sequence getting a view)
csv_data = [list(dict_to_convert[0].keys())]
for dic in dict_to_convert:
csv_data.append([dic[h] for h in dict_to_convert[0].keys()])
else:
csv_file = BytesIO()
csv_data = []
csv_data.append([key.encode('utf-8') for key in dict_to_convert[0].keys()])
for dic in dict_to_convert:
csv_data.append(
[dic[h].encode('utf8')
if isinstance(dic[h], unicode) else str(dic[h]).encode('utf8') # NOQA
for h in dict_to_convert[0].keys()])
except (KeyError, IndexError):
csv_data = []
writer = csv.writer(csv_file)
writer.writerows(csv_data)
csv_file.seek(0)
if six.PY3:
csv_file = BytesIO(csv_file.getvalue().encode('utf-8'))
return csv_file
# base64 encoding that works with both py2 and py3 and yield no warning
base64_encode = base64.encodestring if six.PY2 else base64.encodebytes
class LoginThrottler():
"""Simple login throttler used to limit authentication attempts based on client's ip address.
When using multiple workers, remaining number of attempts can get inconsistent
but will still be limited to num_workers * max_attempts.
"""
def __init__(self, max_attempts=3, delay=1):
self._max_attempts = max_attempts
# Delay in minutes before resetting the attempts counter
self._delay = delay
self._attempts = {}
def get_remaining_attempts(self, ip):
return self._max_attempts - self._attempts.get(ip, [datetime.now(), 0])[1]
def increment_attempts_counter(self, ip):
# Reset all attempt counters when they get hungry for memory
if len(self._attempts) > 10000:
self.__init__()
if self._attempts.get(ip) is None:
# Store first attempt date and number of attempts since
self._attempts[ip] = [datetime.now(), 0]
self._attempts.get(ip)[1] += 1
def is_login_allowed(self, ip):
if self._attempts.get(ip) is None:
return True
# When the delay is expired, reset the counter
if datetime.now() - self._attempts.get(ip)[0] > timedelta(minutes=self._delay):
self.reset(ip)
return True
if self._attempts.get(ip)[1] >= self._max_attempts:
return False
return True
def reset(self, ip):
self._attempts.pop(ip, None)
|
