某大佬仿flask写的web框架
web_frame.py
from werkzeug.local import LocalStack, LocalProxy def get_request_context():
return request_context_stack.top request_context_stack = LocalStack()
request = LocalProxy(get_request_context) class APP(object): def __init__(self):
self.route_map = {}
self.before_request_functions = []
self.after_request_functions = [] def __call__(self, environ, start_response):
return self.wsgi_app(environ, start_response) def route(self, url):
def wrapper(func):
self.route_map[url] = func
return func
return wrapper def before_request(self, func):
def wrapper(*args, **kwargs):
self.before_request_functions.append(func)
return func
return wrapper() def after_request(self, func):
def wrapper(*args, **kwargs):
self.after_request_functions.append(func)
return func
return wrapper() def preprocess_request(self):
for func in self.before_request_functions:
func() def process_response(self):
for func in self.after_request_functions:
func() def wsgi_app(self, environ, start_response):
request_context_stack.push(environ)
url = environ['PATH_INFO']
func = self.route_map.get(url)
if not func:
start_response('404 NOT FOUND', [('Content-type', 'text/plain')])
return ['Page dose not exists!']
self.preprocess_request()
ret = func()
self.process_response()
start_response('200 OK', [('Content-Type', 'text/html')])
return ret def run(self, host='0.0.0.0', port=8888):
from werkzeug.serving import run_simple
run_simple(host, port, self)
app.py
import json
from web_frame import APP
from web_frame import request app = APP() @app.before_request
def test_before_request_1():
print('this is test_before_request_1, ---------------------') @app.before_request
def test_before_request_2():
print('this is test_before_request_2, ---------------------') @app.after_request
def test_after_request_1():
print('this is after_before_request_1, ---------------------') @app.after_request
def test_after_request_2():
print('this is after_before_request_2, ---------------------') @app.route('/a')
def a():
ret = {
'error_code': '',
'msg': 'success',
'data': 'this is a'
}
return json.dumps(ret) @app.route('/b')
def b():
print(request)
ret = {
'error_code': '',
'msg': 'success',
'data': 'this is b'
}
return json.dumps(ret) if __name__ == '__main__':
app.run()
对理解flask的请求流程有一定帮助