转载

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": []
}
原文  https://rollingstarky.github.io/2019/11/28/python-build-rest-api-with-flask-1-modeling/
正文到此结束
Loading...