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
|
# Copyright (C) 2013 Connor Hudson, Tim Radvan
"""Module for writing Snap! extensions.
See example.py for usage.
Handler functions should return unicode. Non-unicode strings are assumed to
be UTF-8 encoded.
"""
__version__ = '0.1.4'
import inspect
try:
from cStringIO import StringIO
except ImportError:
from StringIO import StringIO
from urlparse import urlsplit, parse_qs
from SimpleHTTPServer import SimpleHTTPRequestHandler
from SocketServer import TCPServer
class SnapHandler(SimpleHTTPRequestHandler):
"""An HTTP handler with Flask-style routing."""
routes = {}
special = {
'true': True,
'True': True,
'false': False,
'False': False,
}
@classmethod
def prettify_arg(cls, value):
value = value.decode('utf-8')
value = cls.special.get(value, value)
if isinstance(value, bool):
return value
try:
return int(value)
except ValueError:
try:
return float(value)
except ValueError:
return value # str
def send_head(self):
split_url = urlsplit(self.path)
path = split_url.path
params = parse_qs(split_url.query)
params = dict((k, self.prettify_arg(v[0])) for (k, v) in params.items())
print 'params', params
is_browser = "text/html" in self.headers['Accept']
(status, mime_type, response) = self.get_response(path, params,
is_browser)
if isinstance(response, str):
response = response.decode('utf-8')
else:
response = unicode(response)
if is_browser and mime_type == "text/plain":
mime_type = "text/html"
response = u"""{response}""".format(title=path, response=response)
if isinstance(response, unicode):
response = response.encode('utf-8')
self.send_response(status)
self.send_header("Content-Type", mime_type)
self.send_header("Content-Length", str(len(response)))
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
return StringIO(response)
def get_response(self, path, params, is_browser):
mime_type = "text/plain"
if path in self.routes:
f = self.routes[path]
try:
response = f(**params)
if response is None:
response = ""
elif response is True:
response = "true"
elif response is False:
response = "false"
return (200, mime_type, response)
except KeyboardInterrupt:
raise
except TypeError, e:
for param in inspect.getargspec(f).args:
if param not in params:
response = "ERROR: Missing argument %r" % param
return (400, mime_type, response)
else:
raise
elif path == '/':
return self.index(is_browser)
elif path == '/crossdomain.xml':
return """
<?xml version="1.0"?>
<cross-domain-policy>
<allow-access-from domain="*" to-ports="*"/>
</cross-domain-policy>
"""
else:
return (404, mime_type, "ERROR: Route not found")
def index(self, is_browser):
"""Return the list of routes in plain text format."""
if is_browser:
html = u"""<!DOCTYPE html>
<meta charset="utf8">
<title>/</title>
<ul>
"""
for path in sorted(self.routes):
f = self.routes[path]
params = inspect.getargspec(f).args
qs = "&".join("%s=" % p for p in params)
if qs:
path += "?" + qs
html += u'<li><a href="{path}">{path}</a>'.format(path=path)
return (200, "text/html", html)
else:
response = "\n".join(sorted(self.routes))
return (200, "text/plain", response)
@classmethod
def add_route(self, path, f):
"""Same as the :meth:`route` decorator.
@handler.route('/')
def index():
pass
Is equivalent to:
def index():
pass
handler.add_url_rule('/', 'index', index)
"""
if path in self.routes and path != '/':
raise ValueError, "route already exists"
self.routes[path] = f
@classmethod
def route(self, path, **options):
def decorator(f):
self.add_route(path, f, **options)
return f
return decorator
class Server(TCPServer):
allow_reuse_address = True
def main(handler, port, silent=False):
"""Runs the server for Snap! to connect to."""
httpd = Server(("", port), handler)
if not silent:
print "Serving at port %i" % port
print "Go ahead and launch Snap!"
httpd.serve_forever()
|
