转载

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/

正文到此结束

所属分类：软件架构编程技术

本文标签： build SQLite bug web js 配置 json schema python 模型 update https 安装 App 数据库建模 id mysql git map 代码 ACE 服务端 ip sql GitHub 开发 Word IO 2019 db cat API REST 删除 mapper GMT tab 数据库 RESTful 软件 http ORM 缓存测试 tar 数据插件 ask message session 服务器 UI key
版权声明： 本文为互联网转载文章，出处已在文章中说明(部分除外)。如果侵权，请联系本站长删除，谢谢。
本文海报： 生成海报一生成海报二

其他链接

关于本站

本站定位：个人技术类博客

本站作用：写博客、记日志、闲聊扯淡鼓捣技术。

问题交流

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

二、搭建开发环境

SQLAlchemy

三、数据库建模

序列化

四、Entrypoint

五、测试

热门推荐

相关文章

说给你听

本文目录

随机标签

书籍教程

近期评论

网站信息

其他链接

关于本站

问题交流