封装框架1

Posted 2020-09-08 escapist

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了封装框架1相关的知识，希望对你有一定的参考价值。

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

先掌握框架的核心内容

框架大同小异

WSGI 是Python的标准也需要容器
servlet 是Java的规范需要容器 tomcat

也有的不是WSGI规范例如 tornado

从头实现一个框架 python web框架很简单

已有的框架(这些功能) 通过文档都可以自己学习
从原理上看框架是如何实现的

我们会被大框架的辅助性功能所干扰

最简单的wsgi应用

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

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

return的内容就是 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

get请求不带body 其请求参数通过path或者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 # 解析后的 qs
    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\')
    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\')
    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")

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

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,可以实现在定义视图函数的时候绑定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

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

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)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)
    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)
    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 = []  #VSCode自定义代码片段14——Vue的axios网络请求封装
 VSCode自定义代码片段14——Vue的axios网络请求封装
 VSCode自定义代码片段14——Vue的axios网络请求封装
 VsCode 代码片段-提升研发效率
 text 来自Codyhouse框架的Browserlist片段源代码
 使用实体框架迁移时 SQL Server 连接抛出异常 - 添加代码片段