Python 通过 Flask 框架构建 REST API（一）
2019-11-28 本文已影响0人
rollingstarky
一、REST 简介
REST（Representational State Transfer）是一种软件架构风格或者开发模式，主要面向可以在多种系统之间提供标准的数据通信功能的 Web 服务。
而 REST 风格的 Web 服务则允许请求端通过统一的、预先定义好的无状态行为来访问和操作服务端数据。
下面简要说明 REST 风格的特点。
- Uniform Interface：统一接口即使用 HTTP 提供的一系列方法（或动词，GET、POST 等）操作基于名词的 URI 表示的资源
- Representations：RESTful 服务关注资源以及资源的访问，representation 即机器可读的对资源当前状态的定义。推荐使用 JSON
- Messages：客户端与服务器之间的请求/响应，以及其中包含的元数据（请求头、响应码等）
- Links Between Resources：REST 架构的核心概念即资源，资源可以包含 link 用于驱动多个资源之间的连接和跳转
- Caching：缓存，REST API 中的缓存由 HTTP 头控制
- Stateless：每个请求都必须是独立的，服务器不保存客户端的状态信息；客户端发送的请求必须包含能够让服务器理解请求的所有信息，因此客户端请求可以被任何可用的服务器应答。无状态原则为 REST 服务的可伸缩性（横向扩展）和 Caching 等特性提供了便利。
二、搭建开发环境
virtualenv 创建 Python 虚拟环境：
$ mkdir flask-mysql && cd flask-mysql
$ pip install virtualenv
$ virtualenv venv
$ source venv/bin/activate
虚拟环境中安装 Flask：$ pip install flask
最简单 Flask 应用代码：
from flask import Flask
app = Flask(__name__)
@app.route('/')
def hello_world():
return 'Hello, From Flask!'
使用 $ FLASK_APP=app.py flask run 命令运行上面创建的 app.py 源文件。
SQLAlchemy
Flask 是一个灵活的轻量级 Web 开发框架，它以插件的方式提供了针对各种数据源的交互支持。
这里使用 ORM（Object Relational Mapper）类型的框架 sqlalchemy 作为与数据库交互的工具。安装命令如下：
$ pip install flask-sqlalchemy pymysql
数据库配置代码：
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://<mysql_username>:<mysql_password>@<mysql_host>:<mysql_port>/<mysql_db>'
db = SQLAlchemy(app)
如未安装 Mysql 服务，可以改为使用 Sqlite3 数据库引擎：
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///./<dbname>.db'
三、数据库建模
创建 Authors 模型的代码如下：
class Authors(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(20))
specialisation = db.Column(db.String(50))
def create(self):
db.session.add(self)
db.session.commit()
return self
def __init__(self, name, specialisation):
self.name = name
self.specialisation = specialisation
def __repr__(self):
return '<Author %d>' % self.id
db.create_all()
上述模型代码会在关联的数据库中创建一张名为 authors 的表，同时 sqlalchemy 框架提供的基于模型的 API 也会用来与该数据表进行交互。
模型中的字段定义等同于如下的 SQL 语句：
mysql> show create table authors\G
*************************** 1. row ***************************
Table: authors
Create Table: CREATE TABLE `authors` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`name` varchar(20) DEFAULT NULL,
`specialisation` varchar(50) DEFAULT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=latin1
1 row in set (0.00 sec)
序列化
序列化是指将 Web API 提供的后台数据以特定的形式（如 json 格式）展示给用户，方便前端程序调用。
反序列化则是此过程的逆向操作，将前端发送给 API 的 json 格式的数据持久化到后端数据库中。
这里需要安装 marshmallow 框架将 SQLAlchemy 返回的数据对象转换成 JSON 格式。
$ pip install marshmallow-sqlalchemy
添加如下代码创建 AuthorSchema 序列化器：
from marshmallow_sqlalchemy import ModelSchema
from marshmallow import fields
class AuthorsSchema(ModelSchema):
class Meta(ModelSchema.Meta):
model = Authors
sqla_session = db.session
id = fields.Number(dump_only=True)
name = fields.String(required=True)
specialisation = fields.String(required=True)
四、Entrypoint
创建 /authors entrypoint，响应 GET 方法获取 authors 模型中的所有数据对象并以 JSON 格式返回给用户。
from flask import Flask, request, jsonify, make_response
@app.route('/authors', methods=['GET'])
def index():
get_authors = Authors.query.all()
author_schema = AuthorsSchema(many=True)
authors = author_schema.dump(get_authors)
return make_response(jsonify({"authors": authors}))
以 id 为筛选条件获取某条特定的数据纪录：
@app.route('/authors/<id>', methods=['GET'])
def get_author_by_id(id):
get_author = Authors.query.get(id)
author_schema = AuthorsSchema()
author = author_schema.dump(get_author)
return make_response(jsonify({"author": author}))
依次添加 POST、PUT、DELETE 等方法的响应逻辑，最终代码如下：
from flask import Flask, request, jsonify, make_response
from flask_sqlalchemy import SQLAlchemy
from marshmallow_sqlalchemy import ModelSchema
from marshmallow import fields
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///./test.db'
db = SQLAlchemy(app)
class Authors(db.Model):
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(20))
specialisation = db.Column(db.String(50))
def create(self):
db.session.add(self)
db.session.commit()
return self
def __init__(self, name, specialisation):
self.name = name
self.specialisation = specialisation
def __repr__(self):
return '<Author %d>' % self.id
db.create_all()
class AuthorsSchema(ModelSchema):
class Meta(ModelSchema.Meta):
model = Authors
sqla_session = db.session
id = fields.Number(dump_only=True)
name = fields.String(required=True)
specialisation = fields.String(required=True)
@app.route('/authors', methods=['GET'])
def index():
get_authors = Authors.query.all()
author_schema = AuthorsSchema(many=True)
authors = author_schema.dump(get_authors)
return make_response(jsonify({"authors": authors}))
@app.route('/authors/<id>', methods=['GET'])
def get_author_by_id(id):
get_author = Authors.query.get(id)
author_schema = AuthorsSchema()
author = author_schema.dump(get_author)
return make_response(jsonify({"author": author}))
@app.route('/authors', methods=['POST'])
def create_author():
data = request.get_json()
author_schema = AuthorsSchema()
author = author_schema.load(data)
result = author_schema.dump(author.create())
return make_response(jsonify({"author": result}), 200)
@app.route('/authors/<id>', methods=['PUT'])
def update_author_by_id(id):
data = request.get_json()
get_author = Authors.query.get(id)
if data.get('specialisation'):
get_author.specialisation = data['specialisation']
if data.get('name'):
get_author.name = data['name']
db.session.add(get_author)
db.session.commit()
author_schema = AuthorsSchema(only=['id', 'name', 'specialisation'])
author = author_schema.dump(get_author)
return make_response(jsonify({"author": author}))
@app.route('/authors/<id>', methods=['DELETE'])
def delete_author_by_id(id):
get_author = Authors.query.get(id)
db.session.delete(get_author)
db.session.commit()
return make_response("", 204)
if __name__ == "__main__":
app.run(host='0.0.0.0', debug=True)
五、测试
运行 python app.py 命令开启 Web 服务，使用 httpie 工具测试 REST API。
POST 方法添加新的数据纪录：
$ http POST 127.0.0.1:5000/authors name="starky" specialisation="Python"
HTTP/1.0 200 OK
Content-Length: 92
Content-Type: application/json
Date: Wed, 27 Nov 2019 02:12:41 GMT
Server: Werkzeug/0.16.0 Python/3.7.4
{
"author": {
"id": 1.0,
"name": "starky",
"specialisation": "Python"
}
}
GET 方法获取数据纪录：
$ http GET 127.0.0.1:5000/authors
{
"authors": [
{
"id": 1.0,
"name": "starky",
"specialisation": "Python"
}
]
}
PUT 方法更新已有的数据纪录：
$ http PUT 127.0.0.1:5000/authors/1 name="skitar" specialisation="Go"
{
"author": {
"id": 1.0,
"name": "skitar",
"specialisation": "Go"
}
}
DELETE 方法删除某条数据：
$ http DELETE 127.0.0.1:5000/authors/1
$ http GET 127.0.0.1:5000/authors
{
"authors": []
}
