后端开发中,我们经常使用web框架来实现各种应用,比如python中的flask,django等,go语言中的gin等。web框架提供了很多现成的工具,大大加快了开发速度。这次,我们将动手实现自己的一个web框架。
当我们在浏览器打开链接发起请求之后发生了什么?
http请求会经过WSGI服务器转发给web框架比如flask,flask处理请求并返回响应。WSGI就相当于中间商,处理客户端和框架之间的信息交换。那么WSGI到底是个啥?
WSGI全称服务器网关接口,你想想,针对每一种服务器都需要实现对应的处理接口是件很麻烦的事,WSGI规定了统一的应用接口实现,具体来说,WSGI规定了application应该实现一个可调用的对象(函数,类,方法或者带__call__
的实例),这个对象应该接受两个位置参数:
同时,该对象需要返回可迭代的响应文本。
为了充分理解WSGI,我们定义一个application,参数为environ和回调函数。
def app(environ, start_response): response_body = b"Hello, World!" status = "200 OK" # 将响应状态和header交给WSGI服务器比如gunicorn start_response(status, headers=[]) return iter([response_body])
当利用诸如gunicorn之类的服务器启动该代码,gunicorn app:app
,打开浏览器就可以看到返回的“hello world”信息。
可以看到,app函数中的回调函数start_response将响应状态和header交给了WSGI服务器。
web框架例如flask的核心就是实现WSGI规范,路由分发,视图渲染等功能,这样我们就不用自己去写相关的模块了。
以flask为例,使用方法如下:
from flask import Flask app = Flask(__name__) @app.route("/home") def hello(): return "hello world" if __name__ = '__main__': app.run()
首先定义了一个全局的app实例,然后在对应的函数上定义路由装饰器,这样不同的路由就分发给不同的函数处理。
为了功能上的考量,我们将application定义为类的形式,新建一个api.py文件,首先实现WSGI。
这里为了方便使用了webob这个库,它将WSGI处理封装成了方便的接口,使用pip install webob
安装。
from webob import Request, Response class API: def __call__(self, environ, start_response): request = Request(environ) response = Response() response.text = "Hello, World!" return response(environ, start_response)
API类中定义了`__call__
内置方法实现。很简单,对吧。
路由是web框架中很重要的一个功能,它将不同的请求转发给不同的处理程序,然后返回处理结果。比如:
对于路由 /home ,和路由/about, 像flask一样,利用装饰器将他们绑定到不同的函数上。
# app.py from api.py import API app = API() @app.route("/home") def home(request, response): response.text = "Hello from the HOME page" @app.route("/about") def about(request, response): response.text = "Hello from the ABOUT page"
这个装饰器是如何实现的?
不同的路由对应不同的handler,应该用字典来存放对吧。这样当新的路由过来之后,直接route.get(path, None) 即可。
class API: def __init__(self): self.routes = {} def route(self, path): def wrapper(handler): self.routes[path] = handler return handler return wrapper ...
如上所示,定义了一个routes字典,然后一个路由装饰器方法,这样就可以使用@app.route("/home") 。
然后需要检查每个过来的request,将其转发到不同的处理函数。
有一个问题,路由有静态的也有动态的,怎么办?
用parse这个库解析请求的path和存在的path,获取动态参数。比如:
>>> from parse import parse >>> result = parse("/people/{name}", "/people/shiniao") >>> print(result.named) {'name': 'shiniao'}
除了动态路由,还要考虑到装饰器是不是可以绑定在类上,比如django。另外如果请求不存在路由,需要返回404。
import inspect from parse import parse from webob import Request, Response class API(object): def __init__(self): # 存放所有路由 self.routes = {} # WSGI要求实现的__call__ def __call__(self, environ, start_response): request = Request(environ) response = self.handle_request(request) # 将响应状态和header交给WSGI服务器比如gunicorn # 同时返回响应正文 return response(environ, start_response) # 找到路由对应的处理对象和参数 def find_handler(self, request_path): for path, handler in self.routes.items(): parse_result = parse(path, request_path) if parse_result is not None: return handler, parse_result.named return None, None # 匹配请求路由,分发到不同处理函数 def handle_request(self, request): response = Response() handler, kwargs = self.find_handler(request.path) if handler is not None: # 如果handler是类的话 if inspect.isclass(handler): # 获取类中定义的方法比如get/post handler = getattr(handler(), request.method.low(), None) # 如果不支持 if handler is None: raise AttributeError("method not allowed.", request.method) handler(request, response, **kwargs) else: self.default_response(response) return response def route(self, path): # if the path already exists if path in self.routes: raise AssertionError("route already exists.") # bind the route path and handler function def wrapper(handler): self.routes[path] = handler return handler return wrapper # 默认响应 def default_response(self, response): response.status_code = 404 response.text = "not found."
新建一个app.py文件,然后定义我们的路由处理函数:
from api import API app = API() @app.route("/home") def home(request, response): response.text = "hello, ding." @app.route("/people/{name}") def people(req, resp, name) resp.text = "hello, {}".format(name)
然后我们启动gunicorn:
gunicorn app:app
。
打开浏览器访问:
http://127.0.0.1:8000/people/shiniao
就能看到返回的信息:hello, shiniao
。
测试下访问重复的路由会不会抛出异常,新建test_ding.py 文件。
使用pytest来测试。
import pytest from api import API @pytest.fixture def api(): return API() def test_basic_route(api): @api.route("/home") def home(req, resp): resp.text = "ding" with pytest.raises(AssertionError): @api.route("/home") def home2(req, resp): resp.text = "ding"
好了,以上就是实现了一个简单的web框架,支持基本的路由转发,动态路由等功能。我一直认为最好的学习方法就是模仿,自己动手去实现不同的轮子,写个解释器啊,web框架啊,小的数据库啊,这些对自己的技术进步都会很有帮助。
另外,新年快乐!