封装框架1

时间：2017-04-05 23:16:10 阅读：290 评论：0 收藏：0 [点我收藏+]

标签：img tuple tin input 解析 oar str format 参数

目的:
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=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=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=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=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=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=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

标签：img tuple tin input 解析 oar str format 参数

原文地址：http://www.cnblogs.com/escapist/p/6642287.html

踩

(0)

评论一句话评论（0）

分享档案

更多>

2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)

周排行