<jupyter_start><jupyter_text>Write Python Web Framework ![web](http://7ktuty.com1.z0.glb.clouddn.com/t.jpg)  WSGI(Web Server Gateway Interface)- [PEP 3333 -- Python Web Server Gateway Interface v1.0.1](https://www.python.org/dev/peps/pep-3333/)- 解决 python web server 乱象 mod_python, CGI, FastCGI, etc.- 描述了web server(Gunicorn, uWSGI等)如何与 web application(flask, django等)交互、web application 如何处理请求- 正是有了 WSGI 规范，我们才能在任意 http web server 上跑各种 web 应用- The WSGI interface has two sides: the "server" or "gateway" side, and the "application" or "framework" side   WSGI API (Application side)``` pythondef application(environ, start_response)```- application 就是 WSGI app，一个可调用对象- 参数：    - environ: 一个包含 WSGI 环境信息的字典，由 WSGI 服务器提供，常见的 key 有 PATH_INFO，QUERY_STRING 等    - start_response: 生成 WSGI 响应的回调函数，接收两个参数，status 和 headers- 函数返回响应体的可迭代对象  一个简单的 app 示例``` pythondef application(environ, start_response):    status = '200 OK'    headers =         ('Content-Type', 'text/html; charset=utf8')    ]    start_response(status, headers)    return  [b"Hello, World!"]```<jupyter_code>"""我们用 python 内置的 wsgi server 来跑这个应用：(0_demo_wsgi_app.py)"""


# 导入 python内置的 WSGI server
from wsgiref.simple_server import make_server

def application(environ, start_response):
    print(environ)    # 建议打出来这个字典看看有哪些参数
    status = '200 OK'
    headers = [('Content-Type', 'text/html; charset=utf8')]
    start_response(status, headers)
    return [b"<h1>Hello, World!</h1>"]    # WSGI applications return iterables that yield bytes

if __name__ == '__main__':
    httpd = make_server('127.0.0.1', 8000, application)
    httpd.serve_forever()<jupyter_output><empty_output><jupyter_text>make_server 如何工作```pythondef make_server(    host, port, app,  server_class=WSGIServer,    handler_class=WSGIRequestHandler):    """Create a new WSGI server"""    server = server_class(        (host, port), handler_class    )    server.set_app(app)    return server``` ![继承图](http://7ktuty.com1.z0.glb.clouddn.com/wsgi.png)<jupyter_code># wsgiref.handlers.BaseHandler
    def run(self, application):
        """Invoke the application"""
        # Note to self: don't move the close()!  Asynchronous servers shouldn't
        # call close() from finish_response(), so if you close() anywhere but
        # the double-error branch here, you'll break asynchronous servers by
        # prematurely closing.  Async servers must return from 'run()' without
        # closing if there might still be output to iterate over.
        try:
            self.setup_environ()
            self.result = application(self.environ, self.start_response)
            self.finish_response()
        except:
            try:
                self.handle_error()
            except:
                # If we get an error handling an error, just give up already!
                self.close()
                raise   # ...and let the actual server figure it out.<jupyter_output><empty_output><jupyter_text>常用的 environ | Key            | Contents                                                                    ||----------------|-----------------------------------------------------------------------------|| REQUEST_METHOD | 请求方法                                                                    || PATH_INFO      | 请求路径，比如 /foo/bar/                                                    || QUERY_STRING   | GET 请求参数，比如 foo=bar&bar=spam，我们可以从这个变量中获取用户的请求参数 || HTTP_{HEADER}  | http 头信息，比如 HTTP_HOST 等                                              || wsgi.input     | 包含请求内容的类文件对象(file-like object)，比如 post 请求数据              |   start_response 可调用对象``` pythonstart_response(status, headers)"""status: 一个包含 http 状态码和描述的字符串, 比如 '200 OK'headers: 一个包含 http 头的元祖列表，[('Content-Type', 'text/html; charset=utf8')]"""``` 最后 WSGI 应用返回一个可迭代的 bytes 序列，比如``` python 注意返回的 bytes 编码要符合你指定的返回头def app(environ, start_response):     ...    return [b'hello', b'world']def app(environ, start_response):     ...    yield b'hello'    yield b'world'```  使用 environ 获取查询参数 到这里我们就知道如何编写一个最简单的 WSGI 应用了，我们做个简单的练习，当用户访问 http://localhost:8000/?name=John 的时候， 在网页上输出 "Hello John"。代码如下：<jupyter_code>from wsgiref.simple_server import make_server


def application(environ, start_response):
    # print(environ)
    status = '200 OK'
    headers = [('Content-Type', 'text/html; charset=utf8')]

    query_string = environ['QUERY_STRING']    # 这里是 "name=John"
    name = query_string.split("=")[1]    # 从查询字符串 "name=John" 里获取 "John"
    start_response(status, headers)
    return [b"<h1>Hello, {}!</h1>".format(name)]


if __name__ == '__main__':
    httpd = make_server('127.0.0.1', 8000, application)
    httpd.serve_forever()<jupyter_output><empty_output><jupyter_text>抽象 Request 和 Response 对象 前面看到我们可以直接编写符合 WSGI 规范的应用，但是要做的工作量比较多，比如我们得直接去处理 query string，大部分 web 框架会抽象出 Request/Response 对象，这样一个 web 应用看起来会像这样：``` pythonfrom somewhere import Responsedef application(request):     ...    return Response('blablabla')``` 这样做的好处就是概念上更加清晰，测试更加容易，大部分的 web 框架都采用了类似抽象。接下来我们看看 web 框架是如何映射 WSGI 到 Request/Response 对象的。代码结构将类似这样：``` pythondef request_response_application(function):    def application(environ, start_response):     ...    return application@request_response_applicationdef application(request):     ...    return Response(...)```  实现 Request 实现思路：把 environ 作为构造函数的参数传过去，这样我们就能利用各种子函数来获取我们需要的值。 比如请求地址(HTTP_HOST)，请求参数(QUERY_STRING) 等。比如我们可以用一个函数把 QUERY_STRING 字符串解析后作为 请求参数字典返回，这样使用的时候就方便很多。<jupyter_code>## -*- coding: utf-8 -*-

from six.moves import urllib


class Request(object):
    def __init__(self, environ):
        self.environ = environ

    @property
    def args(self):
        """ 把查询参数转成字典形式 """
        get_arguments = urllib.parse.parse_qs(
            self.environ['QUERY_STRING']
        )
        return {k: v[0] for k, v in get_arguments.items()}
    
    @property
    def path(self):
        return self.environ['PATH_INFO']<jupyter_output><empty_output><jupyter_text>实现 Response Response 对象需要返回的内容大概如下:- 返回内容- 状态码- 字符编码- 返回类型<jupyter_code>import http.client
from six.moves import urllib
from wsgiref.headers import Headers

class Response(object):
    def __init__(self, response=None, status=200, 
                 charset='utf-8', content_type='text/html'):
        self.response = [] if response is None else response
        self.charset = charset
        self.headers = Headers()
        content_type = '{content_type}; charset={charset}'.format(
            content_type=content_type, charset=charset)
        self.headers.add_header('content-type', content_type)
        self._status = status

    @property
    def status(self):
        status_string = http.client.responses.get(self._status, 'UNKNOWN')
        return '{status} {status_string}'.format(
            status=self._status, status_string=status_string)

    def __iter__(self):
        for val in self.response:
            if isinstance(val, bytes):
                yield val
            else:
                yield val.encode(self.charset)<jupyter_output><empty_output><jupyter_text>转换函数现在 Request/Response 对象都有了，还差一个转换函数，用来把之前的 WSGI 函数转换成使用我们的 Request/Response 对象的函数，我们写个装饰器来实现这个功能:``` pythondef request_response_application(func):    def application(environ, start_response):        request = Request(environ)        response = func(request)        start_response(            response.status,            response.headers.items()        )        return iter(response)    return application```<jupyter_code># 试试结合了 Resquest 和 Response 的新 application:(1_request_response_demo.py)

@request_response_application
def application(request):
    name = request.args.get('name', 'default_name')    # 获取查询字符串中的 name
    return Response(['<h1>hello {name}</h1>'.format(name=name)])

if __name__ == '__main__':
    httpd = make_server('127.0.0.1', 8000, application)
    httpd.serve_forever()<jupyter_output><empty_output><jupyter_text>使用 Werkzeug 其实如果你用过 flask 你一定知道 Werkzeug，一套 flask 依赖的 WSGI 工具集。我们换用 Werkzeug 来写上面的应用几乎一样:``` pythonfrom werkzeug.wrappers import Request, Response@Request.applicationdef application(request):    name = request.args.get('name', 'PyCon')    return Response(['hello {name}'.format(name=name)])```   路由 Router 大部分 web 框架实现了基于匹配的路由，将 url 模式与一个可调用对象匹配，比如 flask 的路由方式:``` python@app.route("/user//photos/")def detail(username, photo_id):     ...```<jupyter_code># Router 实现方式如下，维护一个请求路径到可调用对象的 tuple 列表，
# 每次从列表中查找请求路径是否满足当前 pattern， 
# 满足则调用当前 pattern 对应的可调用对象进行处理。否则抛个异常返回 404 response

class NotFoundError(Exception):
    """ url pattern not found """
    pass


class Router(object):
    def __init__(self):
        self.routing_table = []    # 保存 url pattern 和 可调用对象

    def add_route(self, pattern, callback):
        self.routing_table.append((pattern, callback))

    def match(self, path):
        for (pattern, callback) in self.routing_table:
            m = re.match(pattern, path)
            if m:
                return (callback, m.groups())
        raise NotFoundError()


def hello(request, name):
    return Response("<h1>Hello, {name}</h1>".format(name=name))


def goodbye(request, name):
    return Response("<h1>Goodbye, {name}</h1>".format(name=name))


routers = Router()
routers.add_route(r'/hello/(.*)/$', hello)
routers.add_route(r'/goodbye/(.*)/$', goodbye)

def application(environ, start_response):
    try:
        request = Request(environ)
        callback, args = routers.match(request.path)
        response = callback(request, *args)
    except NotFoundError:
        response = Response("<h1>Not found</h1>", status=404)
    start_response(response.status, response.headers.items())
    return iter(response)

# 当然了，如果喜欢类似 flask 那样的装饰器实现，我们也可以使用类的 __call__ 方法(2_router_app.py):
class DecoratorRouter(object):
    def __init__(self):
        self.routing_table = []    # 保存 url pattern 和 可调用对象

    def match(self, path):
        for (pattern, callback) in self.routing_table:
            m = re.match(pattern, path)
            if m:
                return (callback, m.groups())
        raise NotFoundError()

    def __call__(self, pattern):
        def _(func):
            self.routing_table.append((pattern, func))
        return _

routers = DecoratorRouter()

@routers(r'/hello/(.*)/$')
def hello(request, name):
    return Response("<h1>Hello, {name}</h1>".format(name=name))


@routers(r'/goodbye/(.*)/$')
def goodbye(request, name):
    return Response("<h1>Goodbye, {name}</h1>".format(name=name))
""" App 改成 class(3_app_class.py)
只要实现 `__call__` 方法我们就可以把 application 改成类了:
"""

class Application(object):

    def __init__(self, routers, **kwargs):
        self.routers = routers
        
    def __call__(self, environ, start_response):
        try:
            request = Request(environ)
            callback, args = routers.match(request.path)
            response = callback(request, *args)
        except NotFoundError:
            response = Response("<h1>Not found</h1>", status=404)
        start_response(response.status, response.headers.items())
        return iter(response)


application = Application(routers)


# 实现 App 中间件 (4_app_middleware.py)
# 有时候希望在请求之前或请求之后做一些通用的操作，比如记录日志、修改返回结果等
# 可以通过中间件解决。比如我们想把所有的返回结果变成大写：

class UppercaseMiddleware(object):
    
    def __init__(self, app):
        self.app = app

    def __call__(self, environ, start_response):
        for data in self.app(environ, start_response):
            yield data.upper()
                
application = UppercaseMiddleware(Application(routers))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1269452701521245, 'fraction_numerical': 0.00926735443259311, 'mean_word_length': 3.185510428100988, 'pattern_counts': {'":': 0, '<': 47, '<?xml version=': 0, '>': 47, 'https://': 1, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 5, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 9, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '162255', 'license': 'MIT', 'n_tokens_mistral': 4224, 'n_tokens_neox': 4286, 'n_words': 817, 'path': 'Write_python_web_framework.ipynb', 'repo_name': 'PegasusWang/notebooks', 'size': '11438'}