从0到1构建python web框架

框架功能

第一部分: WSGI(gunicorn)/请求处理程序/路由(简单和参数化)

新建项目gunicorn安装

# 新建项目目录
mkdir longfei
cd longfei
python -m venv venv # 创建虚拟环境
source venv/bin/activate # 激活虚拟环境
pip install gunicorn # 安装WSGI服务器

项目目录中, 新建一个init.pyapi.py文件

# /api.py 文件
class API:
    def __call__(self, environ, start_response):
        response_body = b"Hello, World!"
        status = "200 OK"
        start_response(status, headers=[])
        return iter([response_body])

# /init.py 文件
from api import API

app = API()

运行gunicorn init:app访问http://127.0.0.1:8000,可以看见Hello, World!,但现在请求体中的参数在environ变量中,难以解析,我们返回的response也是bytes形式。

请求处理程序

我们可以使用webob库,将environ中的数据转为Request对象,将需要返回的数据转为Response对象,处理起来更加直观方便,直接通过pip安装一下pip install webob

# /api.py 文件
from webob import Request, Response

class API(object):
    def wsgi_app(self, environ, start_response):
        request = Request(environ)
        response = self.handle_request(request)
        return response(environ, start_response)

    # 请求处理程序, 单独方法方便处理数据
    def handle_request(self, request):
        user_agent = request.environ.get("HTTP_USER_AGENT", "No User Agent Found")

        response = Response()
        response.text = f"Hello, my friend with this user agent: {user_agent}"

        return response

    def __call__(self, environ, start_response):
        self.wsgi_app(environ, start_response)

重启你的gunicorn,你应该在浏览器中看到这条新消息。

路由实现

如果想实现不同页面返回不同内容, 那么就要实现路由功能.目前两种主流的路由实现方式见代码

# init.py
from api import API

app = API()

# ===第一种路由实现==================
# 我们希望访问 xxx.com/home 执行方法, 通过装饰器关联
@app.route("/home")
def home(request, response):
    response.text = "Hello from the HOME page"

# 我们希望访问 xxx.com/about 执行方法
@app.route("/about")
def about(request, response):
    response.text = "Hello from the ABOUT page"

# 路由中可以有变量,对应的方法也需要有对应的参数
@app.route("/hello/{name}")
def hello(requst, response, name):
    response.text = f"Hello, {name}"

# ===第二种路由实现===================
# 可以直接通过add_route方法添加添加路由  
def handler1(req, resp):
    resp.text = "handler1"

app.add_route("/handler1", handler1)
# /api.py 文件
from webob import Request, Response
from urllib import parse

class API(object):

    def __init__(self):
        # url路由, [路由:处理器]
        self.routes = {}

    # 路由装饰器程序
    def route(self, path):
        # 添加路由的装饰器
        def wrapper(handler):
            self.add_route(path, handler)
            return handler
        return wrapper

    def add_route(self, path, handler):
        # 相同路径不可重复添加
        assert path not in self.routes, "Such route already exists"
        self.routes[path] = handler

    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 = self.find_handler(request_path=request.path)
        if handler is not None:
            handler(request, response) # 处理器
        else:
            self.default_response(response)
        return response

    def wsgi_app(self, environ, start_response):
        request = Request(environ)
        response = self.handle_request(request)
        return response(environ, start_response)

    def __call__(self, environ, start_response):
        self.wsgi_app(environ, start_response)

第二部分: 检查重复的路径/基于类的处理程序/单元测试

第三部分: 测试客户端/添加路径的替代方式/支持模板

第四部分: 自定义异常处理/支持静态文件/中间件

附录

此处评论已关闭