forked from BreakawayConsulting/webobj
-
Notifications
You must be signed in to change notification settings - Fork 0
/
webobj.py
185 lines (146 loc) · 5.08 KB
/
webobj.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
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
import http.server
import json
import socketserver
from collections import namedtuple
import threading
import inspect
DEFAULT_ADDR = ('localhost', 8080)
def first_matching(check, lst):
return next(filter(check, lst))
class EventStream(threading.Condition):
def __init__(self, web_object):
self.web_object = web_object
lock = threading.Lock()
super().__init__(lock)
class WebObject:
web_fields = []
def __setattr__(self, name, value):
if name not in self.web_fields:
object.__setattr__(self, name, value)
return
self.event_stream.acquire()
object.__setattr__(self, name, value)
self.event_stream.notify_all()
self.event_stream.release()
@property
def web_state(self):
return {fld: getattr(self, fld) for fld in self.web_fields}
@property
def web_data(self):
return json.dumps(self.web_state).encode('utf8')
@property
def event_stream(self):
try:
es = self._event_steam
except AttributeError:
es = EventStream(self)
self._event_steam = es
return es
class Route(namedtuple('Route', ['route', 'content'])):
def matches(self, request):
return request.path == self.route
class Data:
def __init__(self, data):
self.data = data
class File:
def __init__(self, filename):
self.filename = filename
@property
def data(self):
with open(self.filename, 'rb') as f:
return f.read()
class ThreadedServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
pass
class Handler(http.server.BaseHTTPRequestHandler):
routes = property(lambda self: self.server.routes)
def do_error(self, status):
self.send_response(status)
self.end_headers()
self.wfile.write('<html>Error: {}</html>'.format(status).encode())
def handle_one_request(self):
"""Handle a single HTTP request.
"""
try:
self.raw_requestline = self.rfile.readline(65537)
if len(self.raw_requestline) > 65536:
self.requestline = ''
self.request_version = ''
self.command = ''
self.send_error(414)
return
if not self.raw_requestline:
self.close_connection = 1
return
if not self.parse_request():
# An error code has been sent, just exit
return
mname = 'do_' + self.command
if not hasattr(self, mname):
self.send_error(501, "Unsupported method (%r)" % self.command)
return
method = getattr(self, mname)
method()
# actually send the response if not already done.
self.wfile.flush()
except socket.timeout as e:
# a read or a write timed out. Discard this connection
self.log_error("Request timed out: %r", e)
self.close_connection = 1
return
def do_GET(self):
try:
route = first_matching(lambda x: x.matches(self), self.routes)
except StopIteration:
self.do_error(404)
return
content = route.content
if isinstance(content, (Data, File)):
self.send_response(200)
self.end_headers()
self.wfile.write(content.data)
elif isinstance(content, WebObject):
self.send_response(200)
self.end_headers()
self.wfile.write(content.web_data)
elif isinstance(content, EventStream):
self.send_response(200)
self.send_header("Content-type", "text/event-stream")
self.end_headers()
content.acquire()
while True:
json_data = json.dumps(content.web_object.web_state)
data = 'data: {}\n\n'.format(json_data)
try:
self.wfile.write(data.encode('utf-8'))
self.wfile.flush()
except:
content.release()
break
content.wait()
def do_POST(self):
content_length = int(self.headers['Content-Length'])
data = self.rfile.read(content_length)
try:
route = first_matching(lambda x: x.matches(self), self.routes)
except StopIteration:
self.do_error(404)
return
try:
post_args = json.loads(data.decode())
except ValueError:
self.do_error(400)
return
content = route.content
if inspect.ismethod(content):
result = content(**post_args)
self.send_response(200)
self.end_headers()
self.wfile.write(json.dumps({'result': result}).encode('utf=8'))
class Server:
def __init__(self, routes, addr=DEFAULT_ADDR):
self.routes = routes
self.addr = addr
def start(self):
self.server = ThreadedServer(self.addr, Handler)
self.server.routes = self.routes
self.server.serve_forever()