forked from cilame/any-whim
-
Notifications
You must be signed in to change notification settings - Fork 0
/
极简后端服务,快速处理restful风格的接口.py
56 lines (49 loc) · 1.8 KB
/
极简后端服务,快速处理restful风格的接口.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
import re, json
from urllib.parse import parse_qsl
from wsgiref.simple_server import make_server, WSGIServer
import socketserver
class WS(socketserver.ThreadingMixIn, WSGIServer): pass
class Router:
def __init__(self, route):
self.route = route
def parse(self, environ):
pinfo, mthod = environ['PATH_INFO'].rstrip('/'), environ['REQUEST_METHOD']
for uexp, pobj in self.route:
if uexp == pinfo:
f = getattr(pobj, mthod, lambda e:'{} no method {}'.format(pinfo,mthod).encode())
return f(environ) or b''
return b'ERROR: dont match any route.'
def app(environ, start_response, router):
rheaders = [('Content-type', 'application/json'),
('Access-Control-Allow-Origin', '*'),
('Access-Control-Allow-Methods', 'POST'),
('Access-Control-Allow-Headers', 'x-requested-with,content-type'),]
start_response("200 OK", rheaders)
return [router.parse(environ)]
# 按照以下模式增加 class 来处理各种接口请求
class parse_a:
def GET(environ):
query = environ['QUERY_STRING']
# params = dict(parse_qsl(query)) # 将url参数解析成字典
return query.encode()
class parse_b:
def POST(environ):
body = environ['wsgi.input'].read(int(environ.get('CONTENT_LENGTH', 0)))
return body
router = Router([
('/a', parse_a),
('/b', parse_b),
])
def main():
my_server = make_server('', 8000, lambda e,s:app(e,s,router), WS)
my_server.serve_forever()
import threading
threading.Thread(target=main).start()
# 测试代码
import time
time.sleep(1)
from urllib import request
a = request.urlopen('http://127.0.0.1:8000/a?123=333')
b = request.urlopen('http://127.0.0.1:8000/b?123=333',data='asdfasdf'.encode())
print(a.read())
print(b.read())