如何用 flask 优雅的实现 restful api

Posted 2023-04-27

参考技术A 首先，安装Flask
pip install flask
阅读这篇文章之前我假设你已经了解RESTful API的相关概念，如果不清楚，可以阅读我之前写的这篇博客[Designing a RESTful Web API
Flask是一个使用Python开发的基于Werkzeug的Web框架。
Flask非常适合于开发RESTful API，因为它具有以下特点：
?使用Python进行开发，Python简洁易懂
?容易上手
?灵活
?可以部署到不同的环境
?支持RESTful请求分发
我一般是用curl命令进行测试，除此之外，还可以使用Chrome浏览器的postman扩展。
资源
首先，我创建一个完整的应用，支持响应/, /articles以及/article/:id。
from flask import Flask, url_for
app = Flask(__name__)
@app.route('/')
def api_root():
return 'Welcome'
@app.route('/articles')
def api_articles():
return 'List of ' + url_for('api_articles')
@app.route('/articles/<articleid>')
def api_article(articleid):
return 'You are reading ' + articleid
if __name__ == '__main__':
app.run()
可以使用curl命令发送请求：
响应结果分别如下所示：
GET /
Welcome
GET /articles
List of /articles
GET /articles/123
You are reading 123
路由中还可以使用类型定义:
@app.route('/articles/<articleid>')
上面的路由可以替换成下面的例子：
@app.route('/articles/<int:articleid>')
@app.route('/articles/<float:articleid>')
@app.route('/articles/<path:articleid>')
默认的类型为字符串。
请求
请求参数
假设需要响应一个/hello请求，使用get方法，并传递参数name
from flask import request
@app.route('/hello')
def api_hello():
if 'name' in request.args:
return 'Hello ' + request.args['name']
else:
return 'Hello John Doe'
服务器会返回如下响应信息：
GET /hello
Hello John Doe
GET /hello?name=Luis
Hello Luis
请求方法
Flask支持不同的请求方法：
@app.route('/echo', methods = ['GET', 'POST', 'PATCH', 'PUT', 'DELETE'])
def api_echo():
if request.method == 'GET':
return "ECHO: GET\n"
elif request.method == 'POST':
return "ECHO: POST\n"
elif request.method == 'PATCH':
return "ECHO: PACTH\n"
elif request.method == 'PUT':
return "ECHO: PUT\n"
elif request.method == 'DELETE':
return "ECHO: DELETE"
可以使用如下命令进行测试:
curl -X PATCH :5000/echo
不同请求方法的响应如下：
GET /echo
ECHO: GET
POST /ECHO
ECHO: POST
...
请求数据和请求头
通常使用POST方法和PATCH方法的时候，都会发送附加的数据，这些数据的格式可能如下：普通文本(plain text)， JSON，XML，二进制文件或者用户自定义格式。
Flask中使用request.headers类字典对象来获取请求头信息，使用request.data 获取请求数据，如果发送类型是application/json，则可以使用request.get_json()来获取JSON数据。
from flask import json
@app.route('/messages', methods = ['POST'])
def api_message():
if request.headers['Content-Type'] == 'text/plain':
return "Text Message: " + request.data
elif request.headers['Content-Type'] == 'application/json':
return "JSON Message: " + json.dumps(request.json)
elif request.headers['Content-Type'] == 'application/octet-stream':
f = open('./binary', 'wb')
f.write(request.data)
f.close()
return "Binary message written!"
else:
return "415 Unsupported Media Type ;)"
使用如下命令指定请求数据类型进行测试：
curl -H "Content-type: application/json" \
-X POST :5000/messages -d '"message":"Hello Data"'
使用下面的curl命令来发送一个文件:
curl -H "Content-type: application/octet-stream" \
-X POST :5000/messages --data-binary @message.bin
不同数据类型的响应结果如下所示：
POST /messages "message": "Hello Data"
Content-type: application/json
JSON Message: "message": "Hello Data"
POST /message <message.bin>
Content-type: application/octet-stream
Binary message written!
注意Flask可以通过request.files获取上传的文件，curl可以使用-F选项模拟上传文件的过程。
响应
Flask使用Response类处理响应。
from flask import Response
@app.route('/hello', methods = ['GET'])
def api_hello():
data =
'hello' : 'world',
'number' : 3

js = json.dumps(data)
resp = Response(js, status=200, mimetype='application/json')
resp.headers['Link'] = 'http://luisrei.com'
return resp
使用-i选项可以获取响应信息：
curl -i :5000/hello
返回的响应信息如下所示：
GET /hello
HTTP/1.0 200 OK
Content-Type: application/json
Content-Length: 31
Link: http://luisrei.com
Server: Werkzeug/0.8.2 Python/2.7.1
Date: Wed, 25 Apr 2012 16:40:27 GMT
"hello": "world", "number": 3
mimetype指定了响应数据的类型。
上面的过程可以使用Flask提供的一个简便方法实现：
from flask import jsonify
...
# 将下面的代码替换成
resp = Response(js, status=200, mimetype='application/json')
# 这里的代码
resp = jsonify(data)
resp.status_code = 200
状态码和错误处理
如果成功响应的话，状态码为200。对于404错误我们可以这样处理：
@app.errorhandler(404)
def not_found(error=None):
message =
'status': 404,
'message': 'Not Found: ' + request.url,

resp = jsonify(message)
resp.status_code = 404
return resp
@app.route('/users/<userid>', methods = ['GET'])
def api_users(userid):
users = '1':'john', '2':'steve', '3':'bill'
if userid in users:
return jsonify(userid:users[userid])
else:
return not_found()
测试上面的两个URL，结果如下：
GET /users/2
HTTP/1.0 200 OK

"2": "steve"

GET /users/4
HTTP/1.0 404 NOT FOUND

"status": 404,
"message": "Not Found: :5000/users/4"

默认的Flask错误处理可以使用@error_handler修饰器进行覆盖或者使用下面的方法：
app.error_handler_spec[None][404] = not_found
即使API不需要自定义错误信息，最好还是像上面这样做，因为Flask默认返回的错误信息是html格式的。
认证
使用下面的代码可以处理 HTTP Basic Authentication。
from functools import wraps
def check_auth(username, password):
return username == 'admin' and password == 'secret'
def authenticate():
message = 'message': "Authenticate."
resp = jsonify(message)
resp.status_code = 401
resp.headers['WWW-Authenticate'] = 'Basic realm="Example"'
return resp
def requires_auth(f):
@wraps(f)
def decorated(*args, **kwargs):
auth = request.authorization
if not auth:
return authenticate()
elif not check_auth(auth.username, auth.password):
return authenticate()
return f(*args, **kwargs)
return decorated
接下来只需要给路由增加@require_auth修饰器就可以在请求之前进行认证了：
@app.route('/secrets')
@requires_auth
def api_hello():
return "Shhh this is top secret spy stuff!"
现在，如果没有通过认证的话，响应如下所示：
GET /secrets
HTTP/1.0 401 UNAUTHORIZED
WWW-Authenticate: Basic realm="Example"

"message": "Authenticate."

curl通过-u选项来指定HTTP basic authentication，使用-v选项打印请求头：
curl -v -u "admin:secret"
响应结果如下：
GET /secrets Authorization: Basic YWRtaW46c2VjcmV0
Shhh this is top secret spy stuff!
Flask使用MultiDict来存储头部信息，为了给客户端展示不同的认证机制，可以给header添加更多的WWW-Autheticate。
resp.headers['WWW-Authenticate'] = 'Basic realm="Example"'resp.headers.add('WWW-Authenticate', 'Bearer realm="Example"')
调试与日志
通过设置debug=True来开启调试信息：
app.run(debug=True)
使用Python的logging模块可以设置日志信息：
import logging
file_handlewww.huashijixun.com?baiduzhidao.ogging.FileHandler('app.log')
app.logger.addHandler(file_handler)
app.logger.setLevel(logging.INFO)
@app.route('/hello', methods = ['GET'])
def api_hello():
app.logger.info('informing')
app.logger.warning('warning')
app.logger.error('screaming bloody murder!')
return "check your logs\n"
CURL 命令参考
选项
作用
-X 指定HTTP请求方法，如POST，GET
-H 指定请求头，例如Content-type:application/json
-d 指定请求数据
--data-binary 指定发送的文件
-i 显示响应头部信息
-u 指定认证用户名与密码
-v 输出请求头部信息

如何用Python优雅的合并两个Dict

假设有两个dict x和y，合并成一个新的dict，不改变 x和y的值，例如

 
 x = 'a': 1, 'b': 2
 y = 'b': 3, 'c': 4

期望得到一个新的结果Z，如果key相同，则y覆盖x。期望的结果是

>>> z
'a': 1, 'b': 3, 'c': 4

在PEP448中，有个新的语法可以实现，并且在python3.5中支持了该语法，合并代码如下

z = **x, **y

妥妥的一行代码。
由于现在很多人还在用python2，对于python2和python3.0-python3.4的人来说，有一个比较优雅的方法，但是需要两行代码。

z = x.copy()
z.update(y)

上面的方法，y都会覆盖x里的内容，所以最终结果b=3.

不使用python3.5如何一行完成了

如果您还没有使用Python 3.5，或者需要编写向后兼容的代码，并且您希望在单个表达式中运行，则最有效的方法是将其放在一个函数中：

def merge_two_dicts(x, y):
    """Given two dicts, merge them into a new dict as a shallow copy."""
    z = x.copy()
    z.update(y)
    return z

然后一行代码完成调用:

 z = merge_two_dicts(x, y)

你也可以定义一个函数，合并多个dict，例如

def merge_dicts(*dict_args):
    """
    Given any number of dicts, shallow copy and merge into a new dict,
    precedence goes to key value pairs in latter dicts.
    """
    result = 
    for dictionary in dict_args:
        result.update(dictionary)
    return result

然后可以这样使用

z = merge_dicts(a, b, c, d, e, f, g) 

所有这些里面，相同的key，都是后面的覆盖前面的。

一些不够优雅的示范

items

有些人会使用这种方法：

 z = dict(x.items() + y.items())

这其实就是在内存中创建两个列表，再创建第三个列表，拷贝完成后，创建新的dict，删除掉前三个列表。这个方法耗费性能，而且对于python3，这个无法成功执行，因为items()返回是个对象。

>>> c = dict(a.items() + b.items())
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: unsupported operand type(s) for +: 'dict_items' and 
'dict_items'

你必须明确的把它强制转换成list，z = dict(list(x.items()) + list(y.items()))，这太浪费性能了。
另外，想以来于items()返回的list做并集的方法对于python3来说也会失败，而且，并集的方法，导致了重复的key在取值时的不确定，所以，如果你对两个dict合并有优先级的要求，这个方法就彻底不合适了。

>>> x = 'a': []
>>> y = 'b': []
>>> dict(x.items() | y.items())
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: unhashable type: 'list'

这里有一个例子，其中y应该具有优先权，但是由于任意的集合顺序，x的值被保留：

>>> x = 'a': 2
>>> y = 'a': 1
>>> dict(x.items() | y.items())
'a': 2

构造函数

也有人会这么用

z = dict(x, **y)

这样用很好，比前面的两步的方法高效多了，但是可阅读性差，不够pythonic，如果当key不是字符串的时候，python3中还是运行失败

>>> c = dict(a, **b)
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
TypeError: keyword arguments must be strings

Guido van Rossum 大神说了：宣告dict（， 1：3）是非法的，因为毕竟是滥用机制。虽然这个方法比较hacker，但是太投机取巧了。

一些性能较差但是比较优雅的方法

下面这些方法，虽然性能差，但也比items方法好多了。并且支持优先级。

k: v for d in dicts for k, v in d.items()

python2.6中可以这样

 dict((k, v) for d in dicts for k, v in d.items())
 

itertools.chain:

import itertools
z = dict(itertools.chain(x.iteritems(), y.iteritems()))

性能测试

以下是在Ubuntu 14.04上完成的，在Python 2.7（系统Python）中：

>>> min(timeit.repeat(lambda: merge_two_dicts(x, y)))
0.5726828575134277
>>> min(timeit.repeat(lambda: k: v for d in (x, y) for k, v in d.items() ))
1.163769006729126
>>> min(timeit.repeat(lambda: dict(itertools.chain(x.iteritems(),y.iteritems()))))
1.1614501476287842
>>> min(timeit.repeat(lambda: dict((k, v) for d in (x, y) for k, v in d.items())))
2.2345519065856934

在python3.5中

>>> min(timeit.repeat(lambda: **x, **y))
0.4094954460160807
>>> min(timeit.repeat(lambda: merge_two_dicts(x, y)))
0.7881555100320838
>>> min(timeit.repeat(lambda: k: v for d in (x, y) for k, v in d.items() ))
1.4525277839857154
>>> min(timeit.repeat(lambda: dict(itertools.chain(x.items(), y.items()))))
2.3143140770262107
>>> min(timeit.repeat(lambda: dict((k, v) for d in (x, y) for k, v in d.items())))
3.2069112799945287