Python 通过 Flask 框架构建 REST API（一）——数据库建模
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": []
}
正文到此结束
- 本文标签： db cat API REST Word IO 2019 tab 数据库 RESTful 软件 删除 mapper GMT 测试 tar 数据 插件 http ORM 缓存 服务器 UI key ask message session web js 配置 json build SQLite bug update https 安装 App schema python 模型 mysql git map 代码 数据库建模 id ip sql GitHub 开发 ACE 服务端
- 版权声明： 本文为互联网转载文章，出处已在文章中说明(部分除外)。如果侵权，请联系本站长删除，谢谢。
- 本文海报： 生成海报一 生成海报二
热门推荐
相关文章
近期评论
-
文章和留言都翻到11页了 没有OOM
-
-
朋友，翻页到11页，及以后，会出现OOM，无法访问
-
-
搞个gitee的项目
-
-
版本号是多少，你可以下载哪个代码仓库，jdk选1.8 直接跑就行
-
-
-
Loading...
![[HBLOG]公众号](https://www.liuhaihua.cn/img/qrcode_gzh.jpg)
