Uh oh!
There was an error while loading. Please reload this page.
- Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmain.py
More file actions
Latest commit
89 lines (70 loc) · 2.65 KB
/
Copy pathmain.py
File metadata and controls
89 lines (70 loc) · 2.65 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
85
86
87
88
89
importcgi
importjson
fromhttp.serverimportBaseHTTPRequestHandler, HTTPServer
classApp(BaseHTTPRequestHandler):
endpoints= {'get': {}, "post": {}, 'patch': {}, 'put': {}, 'delete': {}}
def_set_headers(self):
self.send_response(200)
self.send_header('Content-type', 'text/json')
length=int(self.headers['Content-Length'])
content=self.rfile.read(length)
temp=str(content).strip('b\'')
self.end_headers()
returntemp
defparse_POST(self):
content_type, pdict=cgi.parse_header(self.headers['content-type'])
ifcontent_type=='multipart/form-data':
post_vars= {}
cg=cgi.FieldStorage(fp=self.rfile, headers=self.headers,
environ={'REQUEST_METHOD': 'POST', 'CONTENT_TYPE': self.headers['Content-Type']})
forkeyincg.keys():
post_vars[key] =cg[key].value
returnpost_vars
@classmethod
defget(cls, endpoint):
defdecorator(handler_func):
cls.endpoints['get'][endpoint] =handler_func
returnhandler_func
returndecorator
@classmethod
defpost(cls, endpoint):
defdecorator(handler_func):
cls.endpoints['post'][endpoint] =handler_func
returnhandler_func
returndecorator
defdo_GET(self):
endpoint=self.path
ifendpointinself.endpoints['get']:
handler_func=self.endpoints['get'][endpoint]
response=handler_func()
self.send_response(200)
self.send_header('Content-type', 'application/json')
self.end_headers()
self.wfile.write(json.dumps(response).encode('utf-8'))
else:
self.send_error(404, 'Endpoint not found')
defdo_POST(self):
endpoint=self.path
self.send_response(200)
self.send_header('Content-type', 'text/plain')
self.end_headers()
ifendpointinself.endpoints['post']:
post_vars=self.parse_POST()
handler_func=self.endpoints['post'][endpoint]
response=handler_func(post_vars)
self.wfile.write(json.dumps(response).encode('utf-8'))
else:
self.send_error(404, 'Endpoint not found')
@App.get('/path')
defmy_func():
return {'message': f'Hello, !'}
@App.post('/path')
defmy_func(payload):
returnpayload
defrun(server_class=HTTPServer, handler_class=App, port=8000):
server_address= ('', port)
httpd=server_class(server_address, handler_class)
print(f'Starting httpd on port {port}...')
httpd.serve_forever()
if__name__=='__main__':
run()