首页 > 代码库 > 封装框架1

封装框架1

目的:
1、学习web框架是如何工作的
2、如何编写用户友好的API 除了功能性能 就是友好

先掌握框架的核心内容

 

最简单的wsgi应用

environ参数:是客户端和服务器端的环境,有容器传递给应用

start_response函数:传递应用要返回的头信息和body体

 

只能是这两个参数是容器与应用通信的  是位置参数,名字可以取别的

def application(environ, start_response):
    start_response(200 ok, [(Content-Type, text/plain)]) # 返回文本格式
    return ["hello,world".encode()] # 返回的要是bytes的可迭代对象


if __name__ == __main__:
    from wsgiref.simple_server import make_server

    server = make_server(0.0.0.0, 4000, app=application) # Python自带的容器
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        server.shutdown()

了解environ参数内容

# 了解environ
def application(environ, start_response):
    for k, v in environ.items():
        print({}--->{}.format(k, v))
    start_response(200 ok, [(Content-Type, text/plain)])
    return ["hello,world".encode()]


if __name__ == __main__:
    from wsgiref.simple_server import make_server

    server = make_server(0.0.0.0, 4000, app=application)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        server.shutdown()

容器是干什么的?
容器是用来做协议转换的
http协议---->转换为WSGI协议

tomcat 也是把http-->servlet 协议 

 

主要关心客户端传来的信息

import os

def application(environ, start_response):
    for k, v in environ.items():
        if k not in os.environ.keys():
            print({}--->{}.format(k, v))
    start_response(200 ok, [(Content-Type, text/plain)])
    return ["hello,world".encode()]
技术分享
REMOTE_ADDR--->127.0.0.1
CONTENT_LENGTH--->
wsgi.version--->(1, 0)
wsgi.multiprocess--->False
wsgi.errors---><_io.TextIOWrapper name=<stderr> mode=w encoding=UTF-8>
REMOTE_HOST--->
SERVER_NAME--->DESKTOP-16F3VS1
wsgi.url_scheme--->http
wsgi.file_wrapper---><class wsgiref.util.FileWrapper>
PATH_INFO--->/favicon.ico
HTTP_PRAGMA--->no-cache
wsgi.run_once--->False
HTTP_CACHE_CONTROL--->no-cache
CONTENT_TYPE--->text/plain
REQUEST_METHOD--->GET
HTTP_ACCEPT_ENCODING--->gzip, deflate, sdch, br
SERVER_SOFTWARE--->WSGIServer/0.2
HTTP_ACCEPT--->image/webp,image/*,*/*;q=0.8
QUERY_STRING--->
HTTP_USER_AGENT--->Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/57.0.2987.98 Safari/537.36
GATEWAY_INTERFACE--->CGI/1.1
wsgi.input---><_io.BufferedReader name=844>
SERVER_PROTOCOL--->HTTP/1.1
SERVER_PORT--->4000
HTTP_CONNECTION--->keep-alive
HTTP_HOST--->127.0.0.1:4000
HTTP_COOKIE--->csrftoken=0SMasfbAKS3PQIfs2p6JBoRfbphIHVF0; _ga=GA1.1.522430974.1483446120
wsgi.multithread--->True
SCRIPT_NAME--->
HTTP_ACCEPT_LANGUAGE--->zh-CN,zh;q=0.8
HTTP_REFERER--->http://127.0.0.1:4000/?name=wanglei&age=23&name=duanyu
View Code

解析QUERY_STRING

from urllib.parse import parse_qs

def application(environ, start_response):
    query_string = environ.get(QUERY_STRING)
    params = parse_qs(query_string)
    name = params.get(name, [unknown])[0]  # 不管是不是单个值得到的值是列表
    print(name)
    start_response(200 ok, [(Content-Type, text/plain)]) # headers
    return ["hello,world {}".format(name).encode()]

http ---> 代表schema
127.0.0.1 ---> 代表host
8000 ----->   代表port
/...  ---->    代表path
?name=wanglei ---> 代表query_string

 

容器:线程/进程管理  生产中gunicorn容器很好     uwsgi容器必须要和Nginx配合才能使用
        协议转换
输入转化为 environ 输出

头信息用start_response返回  body用return 返回

 

可能产生XSS攻击

from urllib.parse import parse_qs

def application(environ, start_response):
    query_string = environ.get(QUERY_STRING)
    params = parse_qs(query_string)
    name = params.get(name, [unknown])[0] 
    print(name)
    start_response(200 ok, [(Content-Type, text/html)]) # 当Content-Type 是 ‘text/html‘ 小心XSS攻击
return ["hello,world {}".format(name).encode()]

url : http://127.0.0.1:4000/?name=<script type=‘text/javascript‘ >alert(‘fuck%20it‘);</script>

技术分享

from urllib.parse import parse_qs
from html import escape

def application(environ, start_response):
    query_string = environ.get(QUERY_STRING)
    params = parse_qs(query_string)
    name = params.get(name, [unknown])[0]
    print(name)
    start_response(200 ok, [(Content-Type, text/html)])
    return ["hello,world {}".format(escape(name)).encode()] # escape函数将字符进行转义

 

webob封装Request对象  要传入environ参数

from webob import Request
from html import escape

def application(environ, start_response):
    request = Request(environ) # Request接受环境作为参数
    params = request.params
    name = params.get(name, unknown) # 默认是最后一个 http://127.0.0.1:4000/?name=wanglei&name=duanyu   
    print(name)
    start_response(200 ok, [(Content-Type, text/html)])
    return ["hello,world {}".format(escape(name)).encode()]  # 结果是 hello,world duanyu

 

webob封装Response对象

 而且不用使用escape来转义字符(内部已经做了处理)

from webob import Request, Response

def application(environ, start_response):
    request = Request(environ)
    params = request.params
    name = params.get(name, unknown)
    print(name)
    response = Response()
    response.text = <h1>hello,world {}</h1>.format(name)  # 设置body体
    response.status_code = 200  # 设置状态码
    response.content_type = text/html  # 设置头信息
    return response(environ, start_response)

设置的头信息  可以不通过start_response里面来设置了     Response对象返回的时候需要传递 environ 和 start_response 参数

 

使用wsgify装饰器

作用: 1. 应用的参数只需要request, 返回Response即可  

        2.返回的时候不用再给Response传递 environ 和start_response 参数了

from webob import Request, Response
from webob.dec import wsgify

@wsgify
def application(request: Request) -> Response:
    params = request.params
    name = params.get(name, unknown)
    print(name)
    response = Response()    # 只需要处理视图函数的最后返回一个 Response对象即可
    response.text = <h1>hello,world {}</h1>.format(name) 
    response.status_code = 200  
    response.content_type = text/html 
    return response

 

实现wsgify装饰器

from webob import Request, Response
from functools import wraps

def wsgify(fn):
    @wraps(fn)
    def wrapper(environ, start_response): # 装饰器中被装饰的函数不需要和装饰的函数参数保持一致
        request = Request(environ)
        response = fn(request)  # 相当于先执行下面的application函数
        return response(environ, start_response)  # 相当于在这里面给 response传参数

    return wrapper


@wsgify
def application(request: Request) -> Response:
    params = request.params
    name = params.get(name, unknown)
    print(name)
    response = Response()
    response.text = <h1>hello,world {}</h1>.format(name)
    response.status_code = 200
    response.content_type = text/html
    return response

 

最基本的路由

from webob import Request, Response
from webob.dec import wsgify

@wsgify
def application(request: Request) -> Response:
    response = Response()
    if request.path == /hello:
        response.text = <h1>hello page</h1>
        response.status_code = 200
        response.content_type = text/html
    elif request.path == /:
        response.text = <h1>index page </h1>
        response.status_code = 200
        response.content_type = text/html
    return response

路由的基本思想

一个URL----->函数   将URL和视图函数分开写

静态路由: 现在开发者只需要关系怎么写应用就可以了

from webob import Request, Response
from webob.dec import wsgify


def hello(request: Request) -> Response:
    response = Response()
    response.text = <h1>hello page </h1>
    response.status_code = 200
    response.content_type = text/html
    return response


def index(request: Request) -> Response:
    return Response(body=<h1>index page</h1>, status=200, content_type=text/html)


routes = {
    /hello: hello,  # 路由和函数映射的字典
    /: index
}


@wsgify
def application(request: Request) -> Response:
    return routes[request.path](request) # --->注意这里要传递参数给视图函数

 

动态路由:

手动添加URL和视图函数的对应关系

from webob import Request, Response
from webob.dec import wsgify


def hello(request: Request) -> Response:
    response = Response()
    response.text = <h1>hello page </h1>
    response.status_code = 200
    response.content_type = text/html
    return response


def index(request: Request) -> Response:
    return Response(body=<h1>index page</h1>, status=200, content_type=text/html)


class Application:
    routes = {}

    @classmethod
    def register(cls, path, handler):
        cls.routes[path] = handler

    @wsgify
    def __call__(self, request):
        return self.routes[request.path](request)


if __name__ == __main__:
    from wsgiref.simple_server import make_server

    Application.register(/hello, hello)  # 注册路由
    Application.register(/, index)
    server = make_server(0.0.0.0, 4000, app=Application()) # 这里要特别注意  当函数调用的时候才会 执行__call__方法体的内容
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        server.shutdown()

 可以实现动态注册路由了  route字典 也就不需要了,不需要事先将对应关系写好了

 

改写动态路由:用类来改写

def index(request: Request) -> Response:
    response = Response(body=hello,world, content_type=text/html, )
    return response


def hello(request: Request) -> Response:
    params = request.params
    name = params.get(name, unknown)
    response = Response()
    response.text = hello,world {}.format(name)
    response.status_code = 200
    response.content_type = text/plain
    print(params)
    return response


class Application: route = {} # 路由和处理函数的对应关系 @classmethod def register(cls, path, handler): # 注册路由函数 cls.route[path] = handler @classmethod def default_handler(cls, request: Request) -> Response: #自定义 找不到路由时的处理方法 return Response(body=Not found, status=404) @wsgify def __call__(self, request: Request) -> Response: return self.route.get(request.path, self.error_handler)(request) # 没有注册的url就返回默认的 handler

webob中已经有了错误的方法 不需要自己编写404错误的函数


  from webob import exc
  @classmethod
  def error_handler(cls, request: Request) -> Response:  # webob 的错误
      raise exc.HTTPNotFound("NOT FOUND")

可以改写为如下,没必要再定义一个错误处理函数,直接出错了就捕获

class Application:
    route = {}

    @classmethod
    def register(cls, path):
        def inner(handler):
            cls.route[path] = handler
            return handler
        return inner

    @wsgify
    def __call__(self, request: Request) -> Response:
        try:
            return self.route[request.path](request)
        except KeyError:
            raise exc.HTTPNotFound(not found .......)

 

我们发现每次都要手动注册url,可以实现在定义视图函数的时候绑定url

class Application:
    route = {}

    @classmethod
    def register(cls, path):  # 用装饰器改写注册方法,实现最便捷的注册路由
        def inner(handler):
            cls.route[path] = handler
            return handler
        return inner

    @wsgify
    def __call__(self, request: Request) -> Response:
        try:
            return self.route[request.path](request)
        except KeyError:
            raise exc.HTTPNotFound(not found .......)

app = Application()


@app.register(/)    # 这样就可以在定义视图的时候绑定url路由了
def index(request: Request) -> Response:
    response = Response(body=hello,world, content_type=text/html, )
    return response


@app.register(/hello)
def hello(request: Request) -> Response:
    params = request.params
    name = params.get(name, unknown)
    response = Response()
    response.text = hello,world {}.format(name)
    response.status_code = 200
    response.content_type = text/plain
    print(params)
    return response

 

使用正则表达式匹配 让路由更加具有表现性

class Application:
    route = {}

    @classmethod
    def register(cls, pattern):
        def inner(handler):
            cls.route[pattern] = handler
            return handler

        return inner

    @wsgify
    def __call__(self, request: Request) -> Response:
        for pattern, handler in self.route.items():
            if re.match(pattern, request.path):
                return handler(request)
        raise exc.HTTPNotFound(NOT FOUND)

app = Application()


@app.register(^/$)  # 因为正则表达式匹配是贪婪模式的 所以需要加上 前后牟定 否则这里会全部匹配
def index(request: Request) -> Response:
    response = Response(body=hello,world, content_type=text/html, )
    return response


@app.register(^/hello$)  # 这里也需要加上牟定
def hello(request: Request) -> Response:
    params = request.params
    name = params.get(name, unknown)
    response = Response()
    response.text = hello,world {}.format(name)
    response.status_code = 200
    response.content_type = text/plain
    print(params)
    return response

 

事先编译好pattern

class Application:
    route = []  # 使用列表 而不是字典

    @classmethod
    def register(cls, pattern):
        def inner(handler):
            cls.route.append((re.compile(pattern), handler))  # 事先编译好
            return handler

        return inner

    @wsgify
    def __call__(self, request: Request) -> Response:
        for pattern, handler in self.route:
            if pattern.match(request.path):
                return handler(request)
        raise exc.HTTPNotFound(NOT FOUND)

app = Application()


@app.register(^/$)
def index(request: Request) -> Response:
    response = Response(body=hello,world, content_type=text/html, )
    return response

 

将正则捕获到的参数 放到 request.args 和 request.kwargs里面(可以使用request.kwargs.name来访问字段)
class _Vars:
    def __init__(self, data=http://www.mamicode.com/None):
        if data is not None:
            self._data = data  # 注意改名
        else:
            self._data = {}

    def __getattr__(self, item):
        try:
            return self._data[item]  # 这样就可以实现request.kwargs.name 这种来访问字段了
        except KeyError:
            raise AttributeError("NO attribute {}".format(item))

    def __setattr__(self, key, value):
        if key != _data:
            raise NotImplemented
        self.__dict__[_data] = value  # 表示不能对他属性进行修改


class Application:
    route = []  # 使用列表 而不是字典

    @classmethod
    def register(cls, pattern):
        def inner(handler):
            cls.route.append((re.compile(pattern), handler))
            return handler

        return inner

    @wsgify
    def __call__(self, request: Request) -> Response:
        for pattern, handler in self.route:
            m = pattern.match(request.path)
            if m:
                request.args = m.groups()
                request.kwargs = _Vars(m.groupdict())
                return handler(request)
        raise exc.HTTPNotFound(NOT FOUND)

app = Application()


@app.register(^/$)
def index(request: Request) -> Response:
    response = Response(body=hello,world, content_type=text/html)
    return response


@app.register(^/hello/(\w+)$)  # 测试 args  http://127.0.0.1:3001/hello/wanglei
def hello(request: Request) -> Response:
    name = request.args[0]
    response = Response()
    response.text = hello,world {}.format(name)
    response.status_code = 200
    response.content_type = text/plain
    return response
@app.register(^/hello/(?P<name>\w+)$)  # 测试 kwargs  http://127.0.0.1:3001/hello/wanglei
def hello(request: Request) -> Response:
    name = request.kwargs.name
    response = Response()
    response.text = hello,world {}.format(name)
    response.status_code = 200
    response.content_type = text/plain
    return response

 

将method也加进匹配规则里面

把 method 也作为匹配的条件
class _Vars:
    def __init__(self, data=http://www.mamicode.com/None):
        if data is not None:
            self._data = data  # 注意改名
        else:
            self._data = {}

    def __getattr__(self, item):
        try:
            return self._data[item]
        except KeyError:
            raise AttributeError("NO attribute {}".format(item))

    def __setattr__(self, key, value):
        if key != _data:
            raise NotImplemented
        self.__dict__[_data] = value


class Application:
    route = []  # 使用列表 而不是字典

    @classmethod
    def register(cls, pattern, method=‘GET‘):  # 只需要在这里加一个 就行了  默认请求方法是GET
        def inner(handler):
            cls.route.append((method, re.compile(pattern), handler))
            return handler

        return inner

    @wsgify
    def __call__(self, request: Request) -> Response:
        for method, pattern, handler in self.route:
            if method.upper() != request.method:
                continue
            m = pattern.match(request.path)
            if m:
                request.args = m.groups()
                request.kwargs = _Vars(m.groupdict())  # 也可以使用命名元组来封装
                return handler(request)
        raise exc.HTTPNotFound(NOT FOUND)

app = Application()


@app.register(^/$)
def index(request: Request) -> Response:
    response = Response(body=hello,world, content_type=text/html, )
    return response


@app.register(^/hello/(?P<name>\w+)$,method=POST)  # 测试 kwargs  http://127.0.0.1:3001/hello/duanyu
def hello(request: Request) -> Response:
    # params = request.params
    name = request.kwargs.name
    response = Response()
    response.text = hello,world {}.format(name)
    response.status_code = 200
    response.content_type = text/plain
    return response

 

实现常见请求方法的注册路由:

请求方法的路由: get put post delete head options 

class _Vars:
    def __init__(self, data=http://www.mamicode.com/None):
        if data is not None:
            self._data = data  # 注意改名
        else:
            self._data = {}

    def __getattr__(self, item):
        try:
            return self._data[item]
        except KeyError:
            raise AttributeError("NO attribute {}".format(item))

    def __setattr__(self, key, value):
        if key != _data:
            raise NotImplemented
        self.__dict__[_data] = value


class Application:
    ROUTE = []  # 使用列表 而不是字典

    @classmethod
    def router(cls, pattern, method=GET):  # 只需要在这里加一个 就行了
        def inner(handler):
            cls.ROUTE.append((method, re.compile(pattern), handler))
            return handler

        return inner

    @classmethod
    def get(cls, pattern):  # 可以通过描述器来实现 不用写7次了
        return cls.router(pattern, method=GET)

    @classmethod
    def post(cls, pattern):
        return cls.router(pattern, method=POST)

    @classmethod
    def put(cls, pattern):
        return cls.router(pattern, method=PUT)

    @classmethod
    def head(cls, pattern):
        return cls.router(pattern, method="HEAD")

    @classmethod
    def delete(cls, pattern):
        return cls.router(pattern, method=DELETE)

    @classmethod
    def options(cls, pattern):  # 可以通过描述器来实现 不用写7次了
        return cls.router(pattern, method=OPTIONS)

    @wsgify
    def __call__(self, request: Request) -> Response:
        for method, pattern, handler in self.ROUTE:
            if method.upper() != request.method:
                continue
            m = pattern.match(request.path)
            if m:
                request.args = m.groups()
                request.kwargs = _Vars(m.groupdict())  # 也可以使用命名元组来封装
                return handler(request)
        raise exc.HTTPNotFound(NOT FOUND)

app = Application()

@app.get(^/$)
def index(request: Request) -> Response:
    response = Response(body=hello,world, content_type=text/html, )
    return response

 

一个url可以有多个请求方法

class _Vars:
    def __init__(self, data=http://www.mamicode.com/None):
        if data is not None:
            self._data = data 
        else:
            self._data = {}

    def __getattr__(self, item):
        try:
            return self._data[item]
        except KeyError:
            raise AttributeError("NO attribute {}".format(item))

    def __setattr__(self, key, value):
        if key != _data:
            raise NotImplemented
        self.__dict__[_data] = value


class Application:
    ROUTE = []  # 使用列表 而不是字典

    @classmethod
    def router(cls, pattern=.*, methods=None):
        def inner(handler):
            cls.ROUTE.append((methods, re.compile(pattern), handler))
            return handler

        return inner

    @classmethod
    def get(cls, pattern): 
        return cls.router(pattern, methods=GET)

    @classmethod
    def post(cls, pattern):
        return cls.router(pattern, methods=POST)

    @classmethod
    def put(cls, pattern):
        return cls.router(pattern, methods=PUT)

    @classmethod
    def head(cls, pattern):
        return cls.router(pattern, methods="HEAD")

    @classmethod
    def delete(cls, pattern):
        return cls.router(pattern, methods=DELETE)

    @classmethod
    def options(cls, pattern):  
        return cls.router(pattern, methods=OPTIONS)

    @wsgify
    def __call__(self, request: Request) -> Response:
        for method, pattern, handler in self.ROUTE:
            if method:  # 如果传递的不是默认值
                if isinstance(method, (list, tuple, set)): # 以列表/元组/集合的方式来指定请求方法
                    if request.method not in method:
                        continue
                elif isinstance(method, str):
                    if request.method != method:
                        continue
            m = pattern.match(request.path)
            if m:
                request.args = m.groups()
                request.kwargs = _Vars(m.groupdict())
                return handler(request)
        raise exc.HTTPNotFound(NOT FOUND)

app = Application()

@app.router(^/$)  # 为了更大限度的自由
def index(request: Request) -> Response:
    response = Response(body=hello,world, content_type=text/html, )
    return response


@app.router(^/hello/(?P<name>\w+)$,methods=(GET, POST))  # 测试 get 和 post 函数是否工作
def hello(request: Request) -> Response:
    # params = request.params
    name = request.kwargs.name
    response = Response()
    response.text = hello,world {}.format(name)
    response.status_code = 200
    response.content_type = text/plain
    return response

对上面的优化

class _Vars:
    def __init__(self, data=http://www.mamicode.com/None):
        if data is not None:
            self._data = data  # 注意改名
        else:
            self._data = {}

    def __getattr__(self, item):
        try:
            return self._data[item]
        except KeyError:
            raise AttributeError("NO attribute {}".format(item))

    def __setattr__(self, key, value):
        if key != _data:
            raise NotImplemented
        self.__dict__[_data] = value


class Application:
    ROUTE = []  # 使用列表 而不是字典

    @classmethod
    def router(cls, pattern=.*, methods=None):
        def inner(handler):
            cls.ROUTE.append((methods, re.compile(pattern), handler))
            return handler

        return inner

    @classmethod
    def get(cls, pattern):  # 可以通过描述器来实现 不用写7次了
        return cls.router(pattern, methods=GET)

    @classmethod
    def post(cls, pattern):
        return cls.router(pattern, methods=POST)

    @classmethod
    def put(cls, pattern):
        return cls.router(pattern, methods=PUT)

    @classmethod
    def head(cls, pattern):
        return cls.router(pattern, methods="HEAD")

    @classmethod
    def delete(cls, pattern):
        return cls.router(pattern, methods=DELETE)

    @classmethod
    def options(cls, pattern):  
        return cls.router(pattern, methods=OPTIONS)

    @wsgify
    def __call__(self, request: Request) -> Response:
        for method, pattern, handler in self.ROUTE:
            if method:  # 如果传递的不是默认值
                if isinstance(method, (list, tuple, set))and request.method not in method:  # 上面这里的判断层次太深了
                    continue
                elif isinstance(method, str) and request.method != method:
                    continue
            m = pattern.match(request.path)
            if m:
                request.args = m.groups()
                request.kwargs = _Vars(m.groupdict())
                return handler(request)
        raise exc.HTTPNotFound(NOT FOUND)

app = Application()

@app.router(^/$)  # 为了更大限度的自由
def index(request: Request) -> Response:
    response = Response(body=hello,world, content_type=text/html, )
    return response


@app.router(^/hello/(?P<name>\w+)$,methods=(GET, POST))  # 测试 get 和 post 函数是否工作
def hello(request: Request) -> Response:
    # params = request.params
    name = request.kwargs.name
    response = Response()
    response.text = hello,world {}.format(name)
    response.status_code = 200
    response.content_type = text/plain
    return response

实现分组路由

class _Vars:
    def __init__(self, data=http://www.mamicode.com/None):
        if data is not None:
            self._data = data
        else:
            self._data = {}

    def __getattr__(self, item):
        try:
            return self._data[item]
        except KeyError:
            raise AttributeError("NO attribute {}".format(item))

    def __setattr__(self, key, value):
        if key != _data:
            raise NotImplemented
        self.__dict__[_data] = value

class Router:
    def __init__(self, prefix=‘‘):
        self.__prefix = prefix.rstrip(/)  
self._routes = [] @property def prefix(self): # 不能修改这个前缀 return self.__prefix def route(self, pattern=.*, method=None): def wrap(handler): self._routes.append((method, re.compile(pattern), handler)) return handler return wrap def get(self, pattern=.*): return self.route(pattern, GET) def put(self, pattern=.*): return self.route(pattern, PUT) def post(self, pattern=.*): return self.route(pattern, POST) def head(self, pattern=.*): return self.route(pattern, HEAD) def delete(self, pattern=.*): return self.route(pattern, DELETE) def options(self, pattern=.*): return self.route(pattern, OPTIONS) def patch(self, pattern=.*): return self.route(pattern, PATCH) def run(self, request: Request): if not request.path.startswith(self.prefix): return for method, pattern, handler in self._routes: if method: # 如果传递的不是默认值 if isinstance(method, (list, tuple, set)) and request.method not in method: continue elif isinstance(method, str) and request.method != method: continue print(method, request.method) m = pattern.match(request.path.replace(self.prefix, ‘‘, 1)) if m: request.args = m.groups() request.kwargs = _Vars(m.groupdict()) return handler(request) class Application: ROUTE = [] @classmethod def register(cls, router: Router): cls.ROUTE.append(router) @wsgify def __call__(self, request: Request) -> Response: for router in self.ROUTE: response = router.run(request) if response: return response raise exc.HTTPNotFound(NOT FOUND) shop = Router(prefix=/shop) @shop.get(^/$) def index(request: Request) -> Response: response = Response(body=hello,world, content_type=text/html, ) return response @shop.get(^/hello/(?P<name>\w+)$) # 测试 get 和 post 函数是否工作 def hello(request: Request) -> Response: name = request.kwargs.name response = Response() response.text = hello,world {}.format(name) response.status_code = 200 response.content_type = text/plain return response Application.register(router=shop) if __name__ == __main__: from wsgiref.simple_server import make_server server = make_server(0.0.0.0, 3001, Application()) server.serve_forever()

 

封装框架1