共 10 篇文章

Vue + Flask 实战开发系列（四）

Vue + Flask 实战开发系列（五）

Vue + Flask 实战开发系列（六）

Vue + Flask 实战开发系列（七）

Vue + Flask 实战开发系列（八）

Vue + Flask 实战开发系列（九）

Vue + Flask 实战开发系列（十）

清单首页Vue+Flask实战开发文章详情

清单「Vue+Flask实战开发」 03/10

Vue + Flask 实战开发系列（三）

TalkPython

通过前面两篇文章，我们已经初步实现了一些简单的接口。还有很多需要做的工作，比如项目结构优化，接口请求权限控制等等。接下来，首先来优化一下，我们的项目结构。前面我们的所有程序，都是写在一个文件中的，这显然是不合理的。这次内容中，我们将使用 Flask 的 Blueprint 功能，完成项目结构的改进和优化。Blueprint 对于大型应用程序非常有价值，可以简化大型应用程序的工作。这次内容有涉及三个方面。第一，完善项目结构；第二、重构 author 接口接口；第三、新增 books 相关接口。

项目结构

以下面是最终完成的项目结构。在结构调整过程中，我重构了 author 接口、新增了 books 接口、加了日志、配置文件等相关内容。

|-- app
|   |-- __init__.py
|   |-- author
|   |   |-- __init__.py
|   |   |-- models.py
|   |   |-- routes.py
|   |   `-- schema.py
|   |-- books
|   |   |-- __init__.py
|   |   |-- models.py
|   |   |-- routes.py
|   |   `-- schema.py
|   `-- utils
|       |-- __init__.py
|       |-- log.py
|       `-- responses.py
|-- config.py
|-- logs
|-- main.py

首先，我们需要在 api 目录下，新建一个名字为 app 的 python 包。然后在 app 包目录下，新建 author、books、utils 三个包。增加完三个目录后，返回上一级，继续新建 main.py 和 config.py 文件。

程序入口文件

每个项目都有入口文件或者 main 文件。使用你喜欢的文本编辑器打开 main.py 编写如下代码。

from app import create_app
app = create_app()

if __name__ == '__main__':
	app.run(port=5000,host='0.0.0.0')

配置文件

接下来，打开 config.py 编写配置程序。

import os
import logging
from dotenv import load_dotenv

basedir = os.path.abspath(os.path.dirname(__file__))

# load env
load_dotenv(os.path.join(basedir, '.flaskenv'))

# log dir
log_dir = os.path.join(basedir, os.getenv('LOG_DIR', 'logs'))

class Config(object):
    SQLALCHEMY_DATABASE_URI = 'sqlite:///books.db'
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    LOG_TO_STDOUT = os.environ.get('LOG_TO_STDOUT')
    LOG_LEVEL = logging.INFO

在配置文件中，我配置了数据库连接 URI，日志的级别和目录。项目日后功能涉及的配置代码，都可以写在这里。例如发送邮件相关配置，缓存相关配置等。配置文件编写完成后，接着编写日志相关的功能。

日志功能

日志功能，是一个必要的功能。每一个部署到生产环境的项目，都会有日志功能。这对于程序 bug 的定位和程序的调试是非常必要的，必不可少的一个功能。首先，进入 utils 目录下，新建一个 log.py 文件，在文件中，编写如下程序。

import os
import datetime
from logging.handlers import BaseRotatingHandler

class DayRotatingHandler(BaseRotatingHandler):
    def __init__(self, filename, mode, encoding=None, delay=False):
        self.date = datetime.date.today()
        self.suffix = "%Y-%m-%d.log"
        super(BaseRotatingHandler, self).__init__(filename, mode, encoding, delay)

    def shouldRollover(self, record):
        return self.date != datetime.date.today()

    def doRollover(self):
        if self.stream:
            self.stream.close()
            self.stream = None
        new_log_file = os.path.join(os.path.split(self.baseFilename)[0], datetime.date.today().strftime(self.suffix))
        self.baseFilename = "{}".format(new_log_file)
        self._open()

上面程序主要功能是实现日志按天分割的功能，这个功能可以让我们更加方便的定位应用在使用过程中出现的问题。

统一的接口响应

我们这个项目是前后端分离的应用，通过接口进行前后端通信的。接口的返回信息（响应）需要统一格式，这包含接口出错时的信息。为了实现这一目标，我们需要自己封装一个统一的接口响应方法。具体实现程序如下：

from flask import make_response, jsonify

INVALID_FIELD_NAME_SENT_422 = {
    "http_code": 422,
    "code": "invalidField",
    "message": "Invalid fields found"
}

INVALID_INPUT_422 = {
    "http_code": 422,
    "code": "invalidInput",
    "message": "Invalid input"
}

MISSING_PARAMETERS_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"
}

FORBIDDEN_403 = {
    "http_code": 403,
    "code": "notAuthorized",
    "message": "You are not authorised to execute this."
}
UNAUTHORIZED_401 = {
    "http_code": 401,
    "code": "notAuthorized",
    "message": "Invalid authentication."
}

NOT_FOUND_HANDLER_404 = {
    "http_code": 404,
    "code": "notFound",
    "message": "route not found"
}

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

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

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


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)

上述程序，我们自定义了程序出现错误是的错误信息，统一的接口响应处理程序。完成上述程序的编写之后，紧接着来完成我们在入口程序中定义的 create_app 函数。

Flask App 核心程序

打开 app 目录下的init.py 文件，编写如下程序。

import os
import logging
from logging.handlers import RotatingFileHandler
import datetime
from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow
from flask_cors import CORS
from flask_migrate import Migrate
from config import Config, log_dir
from app.utils.log import DayRotatingHandler
from app.utils import responses as resp
from app.utils.responses import response_with

db = SQLAlchemy()
cors = CORS()
migrate = Migrate()
marshmallow = Marshmallow()


def create_app(config_class=Config):
    app = Flask(__name__)
    app.config.from_object(config_class)

     # 注册插件
    register_plugins(app)

    # 注册蓝图
    register_blueprints(app)

    # 注册日志处理器
    register_logging(app)

    # 注册错误处理函数
    register_errors(app)

    app.logger.info('Flask Rest Api startup')
    return app

def register_logging(app):
    app.logger.name = 'flask_api'
    log_level = app.config.get("LOG_LEVEL", logging.INFO)
    cls_handler = logging.StreamHandler()
    log_file = os.path.join(log_dir, datetime.date.today().strftime("%Y-%m-%d.log"))
    file_handler = DayRotatingHandler(log_file, mode="a", encoding="utf-8")

    logging.basicConfig(level=log_level,
                        format="%(asctime)s %(name)s "
                               "%(filename)s[%(lineno)d] %(funcName)s() %(levelname)s: %(message)s",
                        datefmt="%Y/%m/%d %H:%M:%S",
                        handlers=[cls_handler, file_handler])

    if not app.debug and not app.testing:
        if app.config['LOG_TO_STDOUT']:
            stream_handler = logging.StreamHandler()
            stream_handler.setLevel(logging.INFO)
            app.logger.addHandler(stream_handler)
        else:
            if not os.path.exists('logs'):
                os.mkdir('logs')
            file_handler = RotatingFileHandler(os.path.join(log_dir, 'flask_api.log'),maxBytes=1024 * 1024 * 50, backupCount=5, encoding='utf-8')
            file_handler.setFormatter(logging.Formatter(
                '%(asctime)s %(name)s %(levelname)s: %(message)s '
                '[in %(pathname)s:%(lineno)d]'))

            file_handler.setLevel(logging.INFO)
            app.logger.addHandler(file_handler)

        app.logger.setLevel(logging.INFO)


def register_plugins(app):
    cors.init_app(app, supports_credentials=True)
    db.init_app(app)
    marshmallow.init_app(app)
    migrate.init_app(app, db)


def register_blueprints(app):

    from app.author import author_bp
    app.register_blueprint(author_bp,url_prefix='/api/author')

    from app.books import books_bp
    app.register_blueprint(books_bp,url_prefix='/api/books')



def register_errors(app):

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

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

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

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

上述程序是这个 Flask 项目最重要的部分，这部分程序集成了日志、数据库迁移、跨域、blueprint，还有我们编写的 author、books 接口都注册在这里。后续如果有新用到的 Flask 扩展，新开发的接口，都需要来这里进行注册操作。完成这部分代码的编写，项目结构调整的事，就告一段落了。接下来的重点，是编写项目业务相关的接口。

重构作者信息接口

由于我们项目的结构的调整，之前写好的 author 接口，需要重构一下。首先，打开 app 目录，在该目录下，创建 author 包。继续在 author 包下面添加 models.py、routes.py、schema.py 文件。

$ (venv) cd app
$ (venv) mkdir author && cd author && touch __init__.py
$ (venv) touch models.py routes.py schema.py

models.py 文件中编写 author 模型程序，具体程序如下：

from app import db

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

schema.py 文件中编写对象序列化程序，具体程序如下：

from marshmallow_sqlalchemy import ModelSchema
from marshmallow import fields
from app.author.models import Author
from app.books.schema import BookSchema
from app import db

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'])

routes.py 文件中编写接口处理接口请求相关程序，具体程序如下：

from flask import request
from app import db
from app.author import author_bp
from app.author.models import Author
from app.author.schema import AuthorSchema
from app.utils.responses import response_with
from app.utils import responses as resp

@author_bp.route('/',methods=['POST'])
def create_author():
    try:
        data = request.get_json()
        print(data)
        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_bp.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_bp.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_bp.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_bp.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_bp.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)

最后，在init.py 文件中，编写如下程序：

from flask import Blueprint
author_bp = Blueprint('author_bp',__name__)
from app.author import routes

上述程序，使用到了 Flask 的 Blueprint 功能，这对于构建大型的 Flask 应用非常有用。

实现图书信息接口

books 接口程序与 author 接口实现过程是一样的，这里我直接给出已经完成的所有程序。

books 的 models 程序，完整程序如下：

from app import db
from marshmallow_sqlalchemy import ModelSchema
from marshmallow import fields

class Book(db.Model):
    __tablename__ = '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

books 的 schema 程序，完整程序如下：

from app.books.models import Book
from marshmallow_sqlalchemy import ModelSchema
from marshmallow import fields
from app import db

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()

books 的 routes 程序，完整程序如下：

from flask import request
from app import db
from app.books import books_bp
from app.books.models import Book
from app.books.schema import BookSchema
from app.utils.responses import response_with
from app.utils import responses as resp

@books_bp.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)

@books_bp.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})

@books_bp.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})

@books_bp.route('/<int:id>', methods=['PUT'])
def update_book_detail(id):
    data = request.get_json()
    get_book = Book.query.get_or_404()
    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})


@books_bp.route('/<int:id>', methods=['PATCH'])
def modify_book_detail(id):
    data = request.get_or_404()
    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 = book_schema.dump(get_book)
    return response_with(resp.SUCCESS_200,value={"book":book})


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

books 的 Blueprint 实现，该程序编写在init.py 文件中，完整程序如下：

from flask import Blueprint

books_bp = Blueprint('books_bp',__name__)

from app.books import routes

完成 books 相关代码编写后，我们需要进行数据迁移和运行程序，来测试我们的程序。

运行程序

回到 api 目录下，运行以下命令，完成数据的迁移操作。

$(venv) flask db init
$(venv) flask db migrate

最后，我们执行如下命令运行我们的 Flask 应用，在执行之前，需要修改.flaskenv 文件 FLASK_APP=main.py，之后再运行下面的命令。

$(venv) flask run

如果一切正常，你将会看到类似于下面的输出结果。

 * Serving Flask app "main.py" (lazy loading)
 * Environment: development
 * Debug mode: on
 * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)
 * Restarting with stat
[2020-06-10 21:04:32,742] INFO in __init__: Flask Rest Api startup
2020/06/10 21:04:32 flask_api __init__.py[37] create_app() INFO: Flask Rest Api startup
 * Debugger is active!
 * Debugger PIN: 267-778-190
[2020-06-10 21:04:33,433] INFO in __init__: Flask Rest Api startup
2020/06/10 21:04:33 flask_api __init__.py[37] create_app() INFO: Flask Rest Api startup

这时我们就可以开始测试我们的接口程序了。这次分享全部内容至此完。