Python 通过 Flask 框架构建 REST API（二）——优化项目架构

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

接上文 Python 通过 Flask 框架构建 REST API（一）——数据库建模。
前面介绍了如何通过 Flask 和 marshmallow 框架写一个完整的单页 Web 应用，作为 REST API 实现基本的增删改查功能。
本篇主要介绍在前文的基础上，如何将单页应用合理地组织到一个架构清晰的项目中。标准化的同时也方便日后的维护。

一、环境搭建

见如下代码：

1
2
3
4
5
6
7
8
9
10
# 创建项目文件夹
$ mkdir author-manager && cd author-manager
# 创建 Python 虚拟环境
$ pip install virtualenv
$ virtualenv venv
$ source venv/bin/activate
# 安装依赖库
(venv) $ pip install flask marshmallow-sqlalchemy flask-sqlalchemy
# 创建文件夹存放源代码
$ mkdir src && cd src

二、初始化应用

创建 main.py 源代码文件初始化 Flask 应用：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# main.py
import os
from flask import Flask
from flask import jsonify

app = Flask(__name__)

if os.environ.get('WORK_ENV') == 'PROD':
    app_config = ProductionConfig
elif os.environ.get('WORK_ENV') == 'TEST':
    app_config = TestingConfig
else:
    app_config = DevelopmentConfig

app.config.from_object(app_config)

创建 run.py 文件作为运行 Web 应用的入口：

1
2
3
4
# run.py
from main import app as application
if __name__ == "__main__":
    application.run(port=5000, host="0.0.0.0", use_reloader=False)

Config

创建 api/config 目录，在其中创建空的 __init__.py 和 config.py 文件，编辑 config.py 加入 config 对象：
$ mkdir -p api/config && touch api/config/__init__.py
$ vim api/config/config.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# src/config/config.py
class Config(object):
    DEBUG = False
    TESTING = False
    SQLALCHEMY_TRACK_MODIFICATIONS = False

class ProductionConfig(Config):
#    SQLALCHEMY_DATABASE_URI = <Production DB RUL>
    pass

class DevelopmentConfig(Config):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = "sqlite:///../authors.db"
    SQLALCHEMY_ECHO = False

class TestingConfig(Config):
    TESTING = True
#    SQLALCHEMY_DATABASE_URI = <Testing DB URL>
    SQLALCHEMY_ECHO = False

PS：每个目录中包含的 __init__.py 空文件用来指示该目录中包含有可供其他代码文件导入的 Python 模块

Database

创建 api/utils 文件夹并编辑 database.py 文件：
$ mkdir -p api/utils && touch api/utils/__init__.py
$ vim api/utils/database.py

1
2
3
4
# src/api/utils/database.py
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

将 main.py 改为如下版本以导入 config 和 db 对象：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# main.py
import os
from flask import Flask
from flask import jsonify
from api.config.config import *
from api.utils.database import db

if os.environ.get('WORK_ENV') == 'PROD':
    app_config = ProductionConfig
elif os.environ.Get('WORK_ENV') == 'TEST':
    app_config = TestingConfig
else:
    app_config = DevelopmentConfig

app = Flask(__name__)
app.config.from_object(app_config)
db.init_app(app)

with app.app_context():
    db.create_all()

此时整个项目的目录结构如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
author-manager
├── src
│   ├── api
│   │   ├── __init__.py
│   │   ├── config
│   │   │   ├── __init__.py
│   │   │   └── config.py
│   │   └── utils
│   │       ├── __init__.py
│   │       └── database.py
│   ├── main.py
│   ├── requirements.txt
│   └── run.py
└── venv

三、数据库关系模型

创建 api/models 文件夹，在其中编辑 books.py 文件作为数据库模型：
$ mkdir -p api/models && touch api/models/__init__.py

books 数据表模型：

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
# src/api/models/books.py
from api.utils.database import db
from marshmallow_sqlalchemy import ModelSchema
from marshmallow import fields

class Book(db.Model):
    __talbename__ = 'books'

    id = db.Column(db.Integer, primary_key=True,
            autoincrement=True)
    title = db.Column(db.String(50))
    year = db.Column(db.Integer)
    author_id = db.Column(db.Integer, db.ForeignKey('authors.id'))

    def __init__(self, title, year, author_id=None):
        self.title = title
        self.year = year
        self.author_id = author_id

    def create(self):
        db.session.add(self)
        db.session.commit()
        return self

class BookSchema(ModelSchema):
    class Meta(ModelSchema.Meta):
        model = Book
        sqla_session = db.session

    id = fields.Number(dump_only=True)
    title = fields.String(required=True)
    year = fields.Integer(required=True)
    author_id = fields.Integer()

authors 数据表模型：

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
# src/api/models/authors.py
from api.utils.database import db
from marshmallow_sqlalchemy import ModelSchema
from marshmallow import fields
from api.models.books import BookSchema

class Author(db.Model):
    __tablename__ = 'authors'
    id = db.Column(db.Integer, primary_key=True,
            autoincrement=True)
    first_name = db.Column(db.String(20))
    last_name = db.Column(db.String(20))
    created = db.Column(db.DateTime, server_default=db.func.now())
    books = db.relationship('Book', backref='Author',
            cascade="all, delete-orphan")

    def __init__(self, first_name, last_name, books=[]):
        self.first_name = first_name
        self.last_name = last_name
        self.books = books

    def create(self):
        db.session.add(self)
        db.session.commit()
        return self

class AuthorSchema(ModelSchema):
    class Meta(ModelSchema.Meta):
        model = Author
        sqla_session = db.session

    id = fields.Number(dump_only=True)
    first_name = fields.String(required=True)
    last_name = fields.String(required=True)
    created = fields.String(dump_only=True)
    books = fields.Nested(BookSchema, many=True,
            only=['title', 'year', 'id'])

四、HTTP 标准响应

在 api/utils 目录下创建 responses.py ，作为 REST API 通用的响应格式：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# src/api/utils/responses.py
from flask import make_response, jsonify

def response_with(response, value=None, message=None,
        error=None, headers={}, pagination=None):
    result = {}
    if value is not None:
        result.update(value)

    if response.get('message', None) is not None:
        result.update({'message': response['message']})

    result.update({'code': response['code']})

    if error is not None:
        result.update({'errors': error})

    if pagination is not None:
        result.update({'pagination': pagination})

    headers.update({'Access-Control-Allow-Origin': '*'})
    headers.update({'server': 'Flask REST API'})

    return make_response(jsonify(result), response['http_code'], headers)

在 responses.py 文件的 response_with 函数前面插入如下代码，定义 http_code：

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
# src/api/utils/responses.py
INVALID_FIELD_NAME_SENT_422 = {
    "http_code": 422,
    "code": "invalidField",
    "message": "Invalid fields found"
    }

INVALID_INPUT_422 = {
    "http_code": 422,
    "code": "missingParameter",
    "message": "Missing parameters"
    }

BAD_REQUEST_400 = {
    "http_code": 400,
    "code": "badRequest",
    "message": "Bad request"
    }

SERVER_ERROR_500 = {
    "http_code": 500,
    "code": "serverError",
    "message": "Server error"
    }

SERVER_ERROR_404 = {
    "http_code": 404,
    "code": "notFound",
    "message": "Resource not found"
    }

UNAUTHORIZED_403 = {
    "http_code": 403,
    "code": "notAuthorized",
    "message": "You are not authorized"
    }

SUCCESS_200 = {
    'http_code': 200,
    'code': 'success',
    }

SUCCESS_201 = {
    'http_code': 201,
    'code': 'success',
    }
SUCCESS_204 = {
    'http_code': 204,
    'code': 'success'
    }

在 main.py 中添加如下代码，导入 status code 定义和 response_with 函数：

1
2
3
4
# src/main.py
import api.utils.responses as resp
from api.utils.responses import response_with
import logging

在 main.py 中 db.init_app(app) 前面添加如下代码，引入错误场景下的标准响应格式：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# src/main.py
@app.after_request
def add_header(response):
    return response

@app.errorhandler(400)
def bad_request(e):
    logging.error(e)
    return response_with(resp.BAD_REQUEST_400)

@app.errorhandler(500)
def server_error(e):
    logging.error(e)
    return response_with(resp.SERVER_ERROR_500)

@app.errorhandler(404)
def not_found(e):
    logging.error(e)
    return response_with(resp.SERVER_ERROR_404)

五、API endpoints

接下来创建 REST API 的访问端点及其对应的路由。编辑 api/routes/authors.py 文件，加入对 POST、GET 等方法的响应逻辑。
$ mkdir -p api/routes && touch api/routes/__init__.py

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
# src/api/routes/authors.py
from flask import Blueprint
from flask import request
from api.utils.responses import response_with
from api.utils import responses as resp
from api.models.authors import Author, AuthorSchema
from api.utils.database import db

author_routes = Blueprint("author_routes", __name__)

@author_routes.route('/', methods=['POST'])
def create_author():
    try:
        data = request.get_json()
        author_schema = AuthorSchema()
        author = author_schema.load(data)
        result = author_schema.dump(author.create())
        return response_with(resp.SUCCESS_201, value={"author":
            result})
    except Exception as e:
        print(e)
        return response_with(resp.INVALID_INPUT_422)

@author_routes.route('/', methods=['GET'])
def get_author_list():
    fetched = Author.query.all()
    author_schema = AuthorSchema(many=True, only=['first_name',
        'last_name','id'])
    authors = author_schema.dump(fetched)
    return response_with(resp.SUCCESS_200, value={"authors":
        authors})

@author_routes.route('/<int:author_id>', methods=['GET'])
def get_author_detail(author_id):
    fetched = Author.query.get_or_404(author_id)
    author_schema = AuthorSchema()
    author = author_schema.dump(fetched)
    return response_with(resp.SUCCESS_200, value={"author":
        author})

@author_routes.route('/<int:id>', methods=['PUT'])
def update_author_detail(id):
    data = request.get_json()
    get_author = Author.query.get_or_404(id)
    get_author.first_name = data['first_name']
    get_author.last_name = data['last_name']
    db.session.add(get_author)
    db.session.commit()
    author_schema = AuthorSchema()
    author = author_schema.dump(get_author)
    return response_with(resp.SUCCESS_200, value={"author":
        author})

@author_routes.route('/<int:id>', methods=['PATCH'])
def modify_author_detail(id):
    data = request.get_json()
    get_author = Author.query.get(id)
    if data.get('first_name'):
        get_author.first_name = data['first_name']
    if data.get('last_name'):
        get_author.last_name = data['last_name']
    db.session.add(get_author)
    db.session.commit()
    author_schema = AuthorSchema()
    author = author_schema.dump(get_author)
    return response_with(resp.SUCCESS_200, value={"author":
        author})

@author_routes.route('/<int:id>', methods=['DELETE'])
def delete_author(id):
    get_author = Author.query.get_or_404(id)
    db.session.delete(get_author)
    db.session.commit()
    return response_with(resp.SUCCESS_204)

然后在 main.py 文件中导入上面创建的 author_routes：
from api.routes.authors import author_routes

并加入以下代码（在 ``@app.after_request之前）以完成路由的注册：app.register_blueprint(author_routes, url_prefix=’/api/authors’)

测试

运行 python run.py 启动 Web 服务，使用 httpie 工具进行测试：

1
2
3
4
5
6
7
8
9
10
11
$ http POST 127.0.0.1:5000/api/authors/ first_name=Jack last_name=Sparrow
{
    "author": {
        "books": [],
        "created": "2019-11-29 04:41:46",
        "first_name": "Jack",
        "id": 2.0,
        "last_name": "Sparrow"
    },
    "code": "success"
}

1
2
3
4
5
6
7
8
9
10
11
$ http 127.0.0.1:5000/api/authors/
{
    "authors": [
        {
            "first_name": "Jack",
            "id": 1.0,
            "last_name": "Sparrow"
        },
    ],
    "code": "success"
}
1
2
3
4
5
6
7
8
9
10
11
$ http 127.0.0.1:5000/api/authors/1
{
    "author": {
        "books": [],
        "created": "2019-11-29 03:22:22",
        "first_name": "Jack",
        "id": 1.0,
        "last_name": "Sparrow"
    },
    "code": "success"
}

同样的方式，创建 api/routes/books.py 文件，添加 book_routes 的响应逻辑：

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
# src/api/routes/books.py
from flask import Blueprint, request
from api.utils.responses import response_with
from api.utils import responses as resp
from api.models.books import Book, BookSchema
from api.utils.database import db

book_routes = Blueprint("book_routes", __name__)

@book_routes.route('/', methods=['POST'])
def create_book():
    try:
        data = request.get_json()
        book_schema = BookSchema()
        book = book_schema.load(data)
        result = book_schema.dump(book.create())
        return response_with(resp.SUCCESS_201, value={"book":
            result})
    except Exception as e:
        print(e)
    return response_with(resp.INVALID_INPUT_422)

@book_routes.route('/', methods=['GET'])
def get_book_list():
    fetched = Book.query.all()
    book_schema = BookSchema(many=True, only=['author_id',
        'title', 'year'])
    books = book_schema.dump(fetched)
    return response_with(resp.SUCCESS_200, value={"books": books})

@book_routes.route('/<int:id>', methods=['GET'])
def get_book_detail(id):
    fetched = Book.query.get_or_404(id)
    book_schema = BookSchema()
    books = book_schema.dump(fetched)
    return response_with(resp.SUCCESS_200, value={"books": books})

@book_routes.route('/<int:id>', methods=['PUT'])
def update_book_detail(id):
    data = request.get_json()
    get_book = Book.query.get_or_404(id)
    get_book.title = data['title']
    get_book.year = data['year']
    db.session.add(get_book)
    db.session.commit()
    book_schema = BookSchema()
    book = book_schema.dump(get_book)
    return response_with(resp.SUCCESS_200, value={"book": book})

@book_routes.route('/<int:id>', methods=['PATCH'])
def modify_book_detail(id):
    data = request.get_json()
    get_book = Book.query.get_or_404(id)
    if data.get('title'):
        get_book.title = data['title']
    if data.get('year'):
        get_book.year = data['year']
    db.session.add(get_book)
    db.session.commit()
    book_schema = BookSchema()
    book = book_schema.dump(get_book)
    return response_with(resp.SUCCESS_200, value={"book": book})

@book_routes.route('/<int:id>', methods=['DELETE'])
def delete_book(id):
    get_book = Book.query.get_or_404(id)
    db.session.delete(get_book)
    db.session.commit()
    return response_with(resp.SUCCESS_204)

在 main.py 中导入 book_routes 并使用 app.register_blueprint 方法注册。同时在 main.py 末尾加入 logging 配置代码，最终效果如下：

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
# src/main.py
import os
import sys
import logging
from flask import Flask
from flask import jsonify
from api.config.config import *
from api.utils.database import db
from api.utils.responses import response_with
import api.utils.responses as resp
from api.routes.authors import author_routes
from api.routes.books import book_routes

if os.environ.get('WORK_ENV') == 'PROD':
    app_config = ProductionConfig
elif os.environ.get('WORK_ENV') == 'TEST':
    app_config = TestingConfig
else:
    app_config = DevelopmentConfig

app = Flask(__name__)
app.config.from_object(app_config)

app.register_blueprint(author_routes, url_prefix='/api/authors')
app.register_blueprint(book_routes, url_prefix='/api/books')

@app.after_request
def add_header(response):
    return response

@app.errorhandler(400)
def bad_request(e):
    logging.error(e)
    return response_with(resp.BAD_REQUEST_400)

@app.errorhandler(500)
def server_error(e):
    logging.error(e)
    return response_with(resp.SERVER_ERROR_500)

@app.errorhandler(404)
def not_found(e):
    logging.error(e)
    return response_with(resp.SERVER_ERROR_404)

db.init_app(app)

with app.app_context():
    db.create_all()

logging.basicConfig(
        stream=sys.stdout,
        format='%(asctime)s|%(levelname)s|%(filename)s:%(lineno)s|%(message)s',
        level=logging.DEBUG)

参考资料

Building REST APIs with Flask