15.7. 07.创建一个简单的REST接口

15.7.1. 问题

你想使用一个简单的REST接口通过网络远程控制或访问你的应用程序,但是你又不想自己去安装一个完整的web框架。

15.7.2. 解决方案

构建一个REST风格的接口最简单的方法是创建一个基于WSGI标准(PEP 3333)的很小的库,下面是一个例子:

resty.py

#!/usr/bin/env python
# -*- coding:utf8 -*-
# auther; 18793
# Date:2020/4/1 17:05
# filename: resty.py

import cgi


def notfound_404(environ, start_response):
    start_response('404 Not Found', [('Content-type', 'text/plain')])
    return [b'Not Found']


class PathDispatcher:
    def __init__(self):
        self.pathmap = {}

    def __call__(self, environ, start_response):
        path = environ['PATH_INFO']
        params = cgi.FieldStorage(environ['wsgi.input'],
                                  environ=environ)
        method = environ['REQUEST_METHOD'].lower()
        environ['params'] = {key: params.getvalue(key) for key in params}
        handler = self.pathmap.get((method, path), notfound_404)
        return handler(environ, start_response)

    def register(self, method, path, function):
        self.pathmap[method.lower(), path] = function
        return function

service_REST.py

#!/usr/bin/env python
# -*- coding:utf8 -*-
# auther; 18793
# Date:2020/4/1 17:08
# filename: service_REST.py
import time
import cgi

_hello_resp = '''\
<html>
  <head>
    <title>Hello {name}</title>
  </head>
  <body>
    <h1>Hello {name}!</h1>
  </body>
</html>'''


def hello_world(environ, start_response):
    start_response('200 OK', [('Content-type', 'text/html')])
    params = environ['params']
    resp = _hello_resp.format(name=params.get('name'))
    yield resp.encode('utf-8')


_localtime_resp = '''\

<?xml version="1.0"?>
<time>
    <year>{t.tm_year}</year>
    <month>{t.tm_mon}</month>
    <day>{t.tm_mday}</day>
    <hour>{t.tm_hour}</hour>
    <minute>{t.tm_min}</minute>
    <second>{t.tm_sec}</second>
</time>'''


def localtime(environ, start_response):
    start_response('200 OK', [('Content-type', 'application/xml')])
    resp = _localtime_resp.format(t=time.localtime())
    yield resp.encode('utf-8')


def wsgi_app(environ, start_response):
    pass
    start_response('200 OK', [('Content-type', 'text/plain')])
    resp = []
    resp.append(b'Hello World\n')
    resp.append(b'Goodbye!\n')
    return resp


def wsgi_app1(environ, start_response):
    pass
    start_response('200 OK', [('Content-type', 'text/plain')])
    yield b'Hello World\n'
    yield b'Goodbye!\n'


def application(environ, start_response):
    import json
    result = [{"name": "John Johnson", "street": "Oslo West 555", "age": 33}]
    response_body = json.dumps(result)
    status = '200 OK'
    response_headers = [('Content-Type', 'text/plain'),
                        ('Content-Length', str(len(response_body)))]
    start_response(status, response_headers)
    return [response_body.encode('utf-8')]


if __name__ == '__main__':
    from resty import PathDispatcher
    from wsgiref.simple_server import make_server

    # Create the dispatcher and register functions
    dispatcher = PathDispatcher()
    dispatcher.register('GET', '/hello', hello_world)
    dispatcher.register('GET', '/localtime', localtime)
    dispatcher.register('GET', '/wsgi_app', wsgi_app)
    dispatcher.register('GET', '/wsgi_app1', wsgi_app1)
    dispatcher.register('GET', '/application', application)

    # Launch a basic server
    httpd = make_server('', 8080, dispatcher)
    print('Serving on port 8080...')
    httpd.serve_forever()

client01.py

#!/usr/bin/env python
# -*- coding:utf8 -*-
# auther; 18793
# Date:2020/4/1 17:10
# filename: client01.py

import urllib.request

## 返回渲染过的html
# u = urllib.request.urlopen('http://localhost:8080/hello?name=Guido')
# print(u.read().decode('utf-8'))

## 返回渲染过的html
# u2 = urllib.request.urlopen('http://localhost:8080/localtime')
# print(u2.read().decode('utf-8'))

## 返回字符串
# u_str = urllib.request.urlopen('http://localhost:8080/wsgi_app?')
# print(u_str.read().decode('utf-8'))

## 可以使用 yield
# u_dict = urllib.request.urlopen('http://localhost:8080/wsgi_app1?')
# print(u_dict.read().decode('utf-8'))

## 返回json格式
u_dict = urllib.request.urlopen('http://localhost:8080/application?')
print(u_dict.read().decode('utf-8'))

尽管WSGI程序通常被定义成一个函数,不过你也可以使用类实例来实现,只要它实现了合适的 __call__() 方法。例如:

class WSGIApplication:
    def __init__(self):
        ...
    def __call__(self, environ, start_response)
       ...

我们已经在上面使用这种技术创建 PathDispatcher 类。 这个分发器仅仅只是管理一个字典,将(方法,路径)对映射到处理器函数上面。 当一个请求到来时,它的方法和路径被提取出来,然后被分发到对应的处理器上面去。 另外,任何查询变量会被解析后放到一个字典中,以 environ['params']形式存储。 后面这个步骤太常见,所以建议你在分发器里面完成,这样可以省掉很多重复代码。 使用分发器的时候,你只需简单的创建一个实例,然后通过它注册各种WSGI形式的函数。 编写这些函数应该超级简单了,只要你遵循 start_response() 函数的编写规则, 并且最后返回字节字符串即可。

当编写这种函数的时候还需注意的一点就是对于字符串模板的使用。 没人愿意写那种到处混合着print()函数 、XML和大量格式化操作的代码。 我们上面使用了三引号包含的预先定义好的字符串模板。 这种方式的可以让我们很容易的在以后修改输出格式(只需要修改模板本身,而不用动任何使用它的地方)。

15.7.3. 参考资料

https://python3-cookbook.readthedocs.io/zh_CN/latest/c11/p05_creating_simple_rest_based_interface.html#id2