/
router.py
158 lines (127 loc) · 4.39 KB
/
router.py
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
import sys
import re
from urllib.parse import parse_qs
rule_pattern = re.compile(
r"""(?P<static>[^<]*) #static
<(?:
(?P<type>[a-zA-Z_][a-zA-Z0-9_]*)
\:
)?
(?P<variable>[a-zA-Z_][a-zA-Z0-9_]*) # variable
>
""",
re.VERBOSE,
)
valid_methods = ["GET", "POST", "DELETE" "PUT", "HEAD"]
def parse_rule(rule):
pos = 0
end = len(rule)
while pos < end:
m = rule_pattern.match(rule, pos)
if m is None: # pure static url
break
d = m.groupdict()
_type = d["type"]
_variable = d["variable"]
if d["static"] and len(d["static"]) > 0:
yield None, None, d["static"]
yield _type, _variable, None
pos = m.end()
if pos < end:
r = rule[pos:]
yield None, None, r
class RouterException(Exception):
pass
class Rule(object):
"""Route rule object
"""
def __init__(self, rule, methods):
self.rule = rule
if methods is None:
self.methods = None
else:
self.methods = set([x.upper() for x in methods])
if "HEAD" not in self.methods and "GET" in self.methods:
self.methods.add("HEAD")
for m in self.methods:
if m not in valid_methods:
raise RouterException("Invalid method %s" % m)
self._regex = None
self._trace = [] # for building url
self.variables = set()
self.build_regex()
def build_url(self, values=None):
if values and len(self.variables) > len(values):
raise RouterException(
"Need %d argument to build the URL. Got %d"
% (len(self.variables), len(values))
)
url_parts = []
for _is_dynamic, var in self._trace:
if _is_dynamic:
try:
url_parts.append(str(values[var]))
except KeyError:
raise RouterException("Need argument '%s' to build the URL." % var)
else:
url_parts.append(var)
return str(u"".join(url_parts))
def build_regex(self):
regex_parts = []
for _type, _variable, _static in parse_rule(self.rule):
if _type is None and _static:
regex_parts.append(re.escape(_static))
self._trace.append((False, _static))
elif _variable:
regex_parts.append("(?P<%s>%s)" % (_variable, "[a-zA-Z0-9_]*"))
self._trace.append((True, _variable))
self.variables.add(_variable)
regex = r"^%s$" % (u"".join(regex_parts))
self._regex = re.compile(regex, re.UNICODE)
class Router(object):
"""Router object for request routing.
"""
def __init__(self):
self.rulesMap = {}
def register(self, path, fn, methods):
if not callable(fn):
raise RouterException("Router only accept callable object.")
r = Rule(path, methods)
self.rulesMap[r] = fn
def __call__(self, p, method="GET"):
return self.get(p, method)
def get(self, path, method="GET"):
f, args = self._match_path(path, method=method)
if not f:
return None
return f, args
def _match_path(self, p, method="GET"):
for rule, fn in self.rulesMap.items():
r = rule._regex
m = r.match(p)
if m is None:
continue
if not method.upper() in rule.methods:
raise RouterException(
"Request method %s not allowed in this app." % method
)
args = {}
for a in rule.variables:
args[a] = m.group(a)
if len(args) == 0:
args = None
return fn, args
return None, None
def url_for(self, fn, **kwargs):
if not callable(fn):
raise RouterException("router url_for method only accept callable object.")
for rule, v in self.rulesMap.items():
if v == fn:
if len(rule.variables) > 0:
return rule.build_url(kwargs)
return rule.build_url()
raise RouterException("callable object doesn't matched any routing rule.")
def all_callables(self):
return self.rulesMap.values()
def remove_all_routes(self):
self.rulesMap.clear()