Python 通过 Flask 框架构建 REST API（一）——数据库建模

发表于 | 分类于 | | 阅读次数:
字数统计: 7.5k | 阅读时长 ≈ 0:07

一、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 虚拟环境：

1
2
3
4
$ mkdir flask-mysql && cd flask-mysql
$ pip install virtualenv
$ virtualenv venv
$ source venv/bin/activate

虚拟环境中安装 Flask：$ pip install flask

最简单 Flask 应用代码：

1
2
3
4
5
6
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

数据库配置代码：

1
2
3
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 模型的代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
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 语句：

1
2
3
4
5
6
7
8
9
10
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 序列化器：

1
2
3
4
5
6
7
8
9
10
11
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 格式返回给用户。

1
2
3
4
5
6
7
8
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 为筛选条件获取某条特定的数据纪录：

1
2
3
4
5
6
@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 等方法的响应逻辑，最终代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
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 方法添加新的数据纪录：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
$ 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 方法获取数据纪录：

1
2
3
4
5
6
7
8
9
10
$ http GET 127.0.0.1:5000/authors
{
    "authors": [
        {
            "id": 1.0,
            "name": "starky",
            "specialisation": "Python"
        }
    ]
}

PUT 方法更新已有的数据纪录：

1
2
3
4
5
6
7
8
$ http PUT 127.0.0.1:5000/authors/1 name="skitar" specialisation="Go"
{
    "author": {
        "id": 1.0,
        "name": "skitar",
        "specialisation": "Go"
    }
}

DELETE 方法删除某条数据：

1
2
3
4
5
6
$ http DELETE 127.0.0.1:5000/authors/1

$ http GET 127.0.0.1:5000/authors
{
    "authors": []
}

参考资料

Building REST APIs with Flask