REST( Representational State Transfer )是一种软件架构风格或者开发模式,主要面向可以在多种系统之间提供标准的数据通信功能的 Web 服务。
而 REST 风格的 Web 服务则允许请求端通过 统一 的、预先定义好的 无状态 行为来访问和操作服务端数据。
下面简要说明 REST 风格的特点。
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
源文件。
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)
创建 /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": [] }