flask + rpc提供json-rpc接口

apicloud+vue 客户端APP搭建

功能模块：用户管理、社交模块、果园模块、抽奖模块、直播群聊模块

服务端项目搭建

创建虚拟环境

conda create -n yingming python=3.8
conda activate yingming

安装开发中使用的依赖模块

pip install flask
pip install flask-redis
pip install flask-mysqldb -i https://pypi.douban.com/simple
pip install flask-sqlalchemy -i https://pypi.douban.com/simple
pip install Flask-PyMongo -i https://pypi.douban.com/simple

新建服务端项目根目录yingmingapi，接着在pycharm中打开项目目录yingmingapi编写manage.py启动程序文件

常见的系统启动程序文件名：manage.py / run.py / main.py / home.py / index.py / common.py

创建启动程序

manage.py，代码：

from flask import Flask

app = Flask(__name__)

@app.route('/')
def index():
    return 'index'

if __name__ == '__main__':
    app.run()

manage.py终不能存放大量的开发代码, 在开发中应该体现的是一种分工精神,所以我们可以把flask中各种功能代码进行分类分文件存储.

创建项目目录结构：

服务端项目根目录/
├── application/            # 项目主要逻辑代码保存目录
|   ├── settings/           # 项目配置存储目录
│   │   ├ dev.py            # 开发阶段的配置文件【本地开发者使用】
│   │   ├ pro.py            # 生产阶段的配置文件【线上服务器使用】
|   |   ├ __init__.py       # 项目公共配置文件，相当于django的global_settings.py
│   ├── __init__.py         # 项目初始化入口
├── manage.py               # 项目的终端管理脚本文件

构建全局初始化工厂函数创建app应用对象

把引导整个项目启动的全局初始化代码,保存到application/__init__py,代码:

from flask import Flask


def init_app() -> Flask:
    """用于创建app实例对象并完成初始化过程的工厂函数"""
    # 实例化flask应用对象
    app: Flask = Flask(__name__)
    return app

manage.py,中调用初始化函数,创建app应用实例对象,代码:

from application import Flask, init_app


app: Flask = init_app()


@app.route('/')
def index():
    return 'welcome to yingmingApp'


if __name__ == '__main__':
    app.run(debug=True)

新增git上传的忽略文件列表，.gitignore，代码：

# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class

# C extensions
*.so

# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST

# PyInstaller
#  Usually these files are written by a python script from a template
#  before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec

# Installer logs
pip-log.txt
pip-delete-this-directory.txt

# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/

# Translations
*.mo
*.pot

# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal

# Flask stuff:
instance/
.webassets-cache

# Scrapy stuff:
.scrapy

# Sphinx documentation
docs/_build/

# PyBuilder
.pybuilder/
target/

# Jupyter Notebook
.ipynb_checkpoints

# IPython
profile_default/
ipython_config.py

# pyenv
#   For a library or package, you might want to ignore these files since the code is
#   intended to run in multiple environments; otherwise, check them in:
# .python-version

# pipenv
#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
#   However, in case of collaboration, if having platform-specific dependencies or dependencies
#   having no cross-platform support, pipenv may install dependencies that don't work, or not
#   install all needed dependencies.
#Pipfile.lock

# poetry
#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
#   This is especially recommended for binary packages to ensure reproducibility, and is more
#   commonly ignored for libraries.
#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock

# PEP 582; used by e.g. github.com/David-OConnor/pyflow
__pypackages__/

# Celery stuff
celerybeat-schedule
celerybeat.pid

# SageMath parsed files
*.sage.py

# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/

# Spyder project settings
.spyderproject
.spyproject

# Rope project settings
.ropeproject

# mkdocs documentation
/site

# mypy
.mypy_cache/
.dmypy.json
dmypy.json

# Pyre type checker
.pyre/

# pytype static type analyzer
.pytype/

# Cython debug symbols
cython_debug/

# PyCharm
#  JetBrains specific template is maintainted in a separate JetBrains.gitignore that can
#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
#  and can be added to the global gitignore or merged into this file.  For a more nuclear
#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
.idea/


nginx/ssl_certs/
!.gitkeep

data

记录代码版本

git init
git config user.name "段沛彤"
git config user.email "3276432700@qq.com"
git remote add origin git@gitee.com:duan-peitong/yingmingapp.git
ssh-keygen -C "3276432700@qq.com"
cat ~/.ssh/ras.pub  # 把公钥复制到gitee 「个人设置」->「安全设置」->「SSH公钥」->「添加公钥」
rm -rf .idea
git add .
git commit -m "fix: crate app instance"
git push -u origin master

基于shell脚本启动项目

项目根目录下yingmingapi/command.sh，代码：

#!/usr/bin/env bash
export FLASK_APP="/home/moluo/Desktop/yingmingapi/manage.py"
export FLASK_DEBUG=True

if [ $1 ]; then
  if [ $1 == "run" ]; then
    flask run --host=0.0.0.0 --port=5000
  else
    flask $1
  fi
fi

终端下设置执行权限

cd ~/Desktop/yingmingapi
chmod 755 command.sh
./command.sh run

终端运行效果：

加载项目配置

这节课修改代码结构实现了根据传入的配置路径自动加载配置。（再导入settings或者application时都会自动执行其内部的__init__文件）

在application/utils/config.py中准备加载配置的函数代码：

from flask import Flask


class Config(object):
    """项目配置加载类"""
    def __init__(self, app: Flask = None, config_path: str = None):
        if app:
            self.init_app(app, config_path)

    def init_app(self, app: Flask = None, path: str = None):
        """
        项目配置初始化函数
        :param app: 当前flask应用实例对象[python中的对象属于引用类型，所以函数内部改了app的数据，外界的app也会修改]
        :param path: 配置文件的导包路径   application.settings.dev
        :return:
        """
        # 先加载默认配置settings.__init__，然后加载当前指定配置config_path
        init_path: str = ".".join(path.split(".")[:-1])
        app.config.from_object(init_path)  # 先加载settings.__init__
        app.config.from_object(path)  # 再加载settings.dev或者settings.prod

编写项目默认配置文件, application/settings/__init__.py代码:

"""公共配置"""
# 调试模式
DEBUG: bool = False
# 语言
LANGUAGE: str = "en"
# 时区
TZ: str = "UTC"

当然, 项目开发过程完成以后肯定会项目上线,所以针对配置文件,我们可以准备不同环境下的配置

application/settings/dev.py,代码:

"""本地配置"""
# 调试模式
DEBUG: bool = True
# 语言
LANGUAGE: str = "zh_hans"
# 时区
TZ: str = "Asia/Shanghai"

application/settings/pro.py,代码:

"""生产配置"""
# 调试模式
DEBUG: bool = False
# 语言
LANGUAGE: str = "zh_hans"
# 时区
TZ: str = "UTC"

在项目入口文件application/__init__py中加载配置,代码:

from flask import Flask
from application.utils.config import Config


# 实例化配置加载类
config: Config = Config()


def init_app(config_path: str) -> Flask:
    """用于创建app实例对象并完成初始化过程的工厂函数"""
    # 实例化flask应用对象
    app: Flask = Flask(__name__)
    # 加载配置
    config.init_app(app, config_path)

    return app

在创建app对象的项目启动文件manage.py中,设置配置路径

from application import Flask, init_app

app: Flask = init_app("application.settings.dev")


@app.route('/')
def index():
    return 'welcome to yingmingApp'


if __name__ == '__main__':
    app.run()

提交版本

git add .
git commit -m "fix: add config function, init porject config"
git push -u origin master

数据库初始化

SQLAlchemy初始化

默认项目配置文件中增加配置选项,application/settings/__init__.py,代码:

"""公共配置"""


"""调试模式"""
DEBUG: bool = False

"""本地化国际化"""
# 语言
LANGUAGE: str = "en"


"""数据库配置"""
# 数据库连接
SQLALCHEMY_DATABASE_URI: str = ""

# 动态追踪修改设置
SQLALCHEMY_TRACK_MODIFICATIONS: bool = False

# 查询时会显示原始SQL语句
SQLALCHEMY_ECHO: bool = False

开发配置settings/dev.py中,配置数据库连接信息,代码:

"""本地配置"""
# 调试模式
DEBUG: bool = True
# 语言
LANGUAGE: str = "zh_hans"

"""数据库配置"""
# 数据库连接
SQLALCHEMY_DATABASE_URI: str = "mysql://yingminguser:yingming@127.0.0.1:3306/yingming?charset=utf8mb4"

# 查询时会显示原始SQL语句
SQLALCHEMY_ECHO: bool = True

在mysql终端下, 创建属于当前项目的数据库管理用户, 命令如下:

mysql -uroot -p123456
create database yingming charset=utf8mb4;
# 针对当前数据库配置账户信息
# mysql8.0之后
create user 'yingminguser'@'%' identified with mysql_native_password by 'yingming';
GRANT ALL ON yingming.* TO 'yingminguser'@'%';

# mysql8.0之前
# create user yingminguser identified by 'yingming'; 
# grant all privileges on yingming.* to 'yingminguser'@'%';
# flush privileges;

在项目初始化主程序中,对数据库功能进行初始化,application/__init__.py,代码:

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

from application.utils.config import init_config

# SQLAlchemy初始化
db: SQLAlchemy = SQLAlchemy()

def init_app(config_path: str):
    """用于创建app实例对象并完成初始化过程的工厂函数"""
    app: Flask = Flask(__name__)
    # 加载配置
    init_config(app, config_path)
    # print(app.config)

    # SQLAlchemy加载配置
    db.init_app(app)

    return app

提交版本

git add .
git commit -m "fix: connect mysql"
git push

Redis数据库初始化

默认配置文件，application/settings/__init__.py，代码：

"""公共配置"""


"""调试模式"""
DEBUG: bool = False

"""本地化国际化"""
# 语言
LANGUAGE: str = "en"


"""mysql数据库配置"""
# 数据库连接
SQLALCHEMY_DATABASE_URI: str = ""

# 动态追踪修改设置
SQLALCHEMY_TRACK_MODIFICATIONS: bool = False

# 查询时会显示原始SQL语句
SQLALCHEMY_ECHO: bool = False

"""redis数据库配置"""
REDIS_URL = "redis://@127.0.0.1:6379/0"

开发配置文件，application/settings/dev.py，代码：

"""本地配置"""


"""调试模式"""
DEBUG: bool = True


"""本地化国际化"""
# 语言
LANGUAGE: str = "zh_hans"


"""数据库配置"""
# 数据库连接
SQLALCHEMY_DATABASE_URI: str = "mysql://yingminguser:yingming@127.0.0.1:3306/yingming?charset=utf8mb4"

# 查询时会显示原始SQL语句
SQLALCHEMY_ECHO: bool = True


"""redis配置"""
# 默认缓存数据
REDIS_URL: str = "redis://:@127.0.0.1:6379/0"
# 验证相关缓存
CHECK_URL: str  = "redis://:@127.0.0.1:6379/1"

在项目初始化主程序中, 对redis进行初始化,applicaiton/__init__.py,代码:

# 项目初始化主程序
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_redis import FlaskRedis

from application.utils.config import Config


"""加载组件[单例模式]"""

# 初始化配置加载类
config: Config = Config()

# SQLAlchemy初始化
db: SQLAlchemy = SQLAlchemy()

# redis初始化
redis_cache: FlaskRedis = FlaskRedis(config_prefix="REDIS")
redis_check: FlaskRedis = FlaskRedis(config_prefix="CHECK")


def init_app(config_path: str) -> Flask:
    """用于创建app实例对象并完成初始化过程的工厂函数"""
    # 实例化flask应用对象
    app: Flask = Flask(__name__)

    # 加载配置
    config.init_app(app, config_path)

    # 加载mysql数据库配置
    db.init_app(app)

    # redis加载配置
    redis_cache.init_app(app)
    redis_check.init_app(app)

    return app

mongoDB数据库初始化

在接下来的项目开发中，我们会先使用pymongo进行数据库操作，后面改成MongoEngineORM框架。所以暂时我们先进行pymongo的初始化。

进入mongoDB交互终端，分配一个账号给当前项目

# 如果有mongodb的账户管理员，则直接进入mongoDB终端，通过如下操作创建yingming数据库的管理员
mongosh
use admin
db.auth("root","123456")

# 给yingming数据库创建一个数据库拥有者身份的管理员账号
use yingming
db.createUser({
    user: "yingming",
    pwd: "yingming",
    roles: [
        { role: "dbOwner", db: "yingming"}
    ]
})

application/settings/__init__.py，代码：

"""公共配置"""


"""调试模式"""
DEBUG: bool = False

"""本地化国际化"""
# 语言
LANGUAGE: str = "en"


"""mysql数据库配置"""
# 数据库连接
SQLALCHEMY_DATABASE_URI: str = ""

# 动态追踪修改设置
SQLALCHEMY_TRACK_MODIFICATIONS: bool = False

# 查询时会显示原始SQL语句
SQLALCHEMY_ECHO: bool = False

"""redis数据库配置"""
REDIS_URL: str = "redis://@127.0.0.1:6379/0"


"""mongoDB配置"""
# mongodb://[username:password@]host1[:port1][,...hostN[:portN]][/[defaultauthdb][?options]]
MONGO_URI: str = "mongodb://:@127.0.0.1:27017/admin"

settings/dev.py，代码：

"""本地配置"""


"""调试模式"""
DEBUG: bool = True


"""本地化国际化"""
# 语言
LANGUAGE: str = "zh_hans"


"""数据库配置"""
# 数据库连接
SQLALCHEMY_DATABASE_URI: str = "mysql://yingminguser:yingming@127.0.0.1:3306/yingming?charset=utf8mb4"

# 查询时会显示原始SQL语句
SQLALCHEMY_ECHO: bool = True


"""redis配置"""
# 默认缓存数据
REDIS_URL: str = "redis://:@127.0.0.1:6379/0"
# 验证相关缓存
CHECK_URL: str  = "redis://:@127.0.0.1:6379/1"

"""mongoDB配置"""
MONGO_URI: str = "mongodb://yingming:yingming@127.0.0.1:27017/yingming"

application/__init__.py，对mongoDB进行初始化

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_redis import FlaskRedis
from flask_pymongo import PyMongo

from application.utils.config import init_config

"""加载组件[单例模式]"""
# SQLAlchemy实例化
db: SQLAlchemy = SQLAlchemy()

# redis实例化
redis_cache: FlaskRedis = FlaskRedis(config_prefix="REDIS")
redis_check: FlaskRedis = FlaskRedis(config_prefix="CHECK")

# mongoDB实例化
mongo: PyMongo = PyMongo()


def init_app(config_path: str) -> Flask:
    """用于创建app实例对象并完成初始化过程的工厂函数"""
    app: Flask = Flask(__name__)

    # 初始化项目配置
    init_config(app=app, config_path=config_path)

    # SQLAlchemy加载配置
    db.init_app(app)

    # redis加载配置
    redis_cache.init_app(app)
    redis_check.init_app(app)

    # pymongo加载配置
    mongo.init_app(app)

    return app

提交版本

git add .
git commit -m "fix: add redis configration and mongodb configration"
git push

完成上面的项目构建步骤以后，此时目录结构新增logs目录，目录结构如下：

yingmingapi/                  # 项目根目录
├── logs/                   # 项目日志存储目录
|   └ yingming.log            # 项目日志
├── application/            # 项目主要逻辑代码保存目录
|   ├── settings/           # 项目配置存储目录
│   │   ├ dev.py            # 开发阶段的配置文件
│   │   ├ pro.py            # 生产阶段的配置文件
|   |   └ __init__.py       # 项目公共配置文件
|   ├── utils/              # 工具函数库/类库
│   │   └ config.py         # 配置相关的辅助类或者辅助函数
│   └── __init__.py         # 项目初始化文件[入口程序，APP工厂函数]
├── manage.py               # 项目的终端管理脚本文件
└── command.sh              # 终端shell命令脚本，用于完成终端操作和启动项目

日志初始化

flask中本身内置了基于loging模块封装的日志功能的,我们在使用的时候, 一般日志如果不是核心重点,则通过由python内置的logging模块进行配置集成使用即可, 如果项目中日志发挥作用比较重要, 则一般安装部署ELK日志分析系统.

日志的等级

从高到低，依次：
	FATAL/CRITICAL = 致命的，危险的
	ERROR = 错误
	WARNING = 警告
	INFO = 信息
	DEBUG = 调试

flask日志功能的基本使用

manage.py

from application import Flask, init_app

app: Flask = init_app("application.settings.dev")


@app.route('/')
def index():
    app.logger.debug("hello, debug")
    app.logger.info("hello, info")
    app.logger.warning("hello, warning")
    app.logger.error("hello, error")
    app.logger.critical("hello, critical")
    return 'welcome to yingmingApp'


if __name__ == '__main__':
    app.run()

在项目运行时，框架本身会不断记录代码程序运行的日志错误，但是我们将来编写的业务逻辑也可能存在抛出异常的情况，这些异常和框架代码本身无关，如果都混在一起的话，对于找出错误，是没有帮助的。所以，我们在这里新建一个日志记录器，专门只是记录我们业务逻辑的相关日志。

构建日志模块

把日志初始化相关的代码封装成一个日志配置类Logger，application/utils/logger.py，代码：

import logging
from logging.handlers import RotatingFileHandler       # 按文件大小分割日志文件
from logging.handlers import TimedRotatingFileHandler    # 按时间片分割日志文件
from flask import Flask

class Logger(object):
    """日志配置类"""
    def __init__(self, app: Flask = None):
        """
        日志实例化
        :param app: 当前flask应用实例对象
        """
        if app:
            self.init_app(app)

    def init_app(self, app: Flask = None)-> None:
        """
        读取项目的日志配置项
        :param app: 当前flask应用实例对象
        :return: None
        """
        self.app = app
        self.log_file = self.app.BASE_DIR / self.app.config.get("LOG_FILE", 'logs/app.log')
        self.log_level = self.app.config.get("LOG_LEVEL", 'INFO')
        self.log_backpu_count = self.app.config.get("LOG_BACKPU_COUNT", 31)
        self.log_format = self.app.config.get("LOG_FORMAT", '%(asctime)s - %(levelname)s - %(filename)s - %(funcName)s - %(lineno)s - %(message)s')
        self.log_rotating_time = self.app.config.get("LOG_ROTATING_TIME", "midnight")
        self.log_charseter = self.app.config.get("LOG_charseter", 'UTF-8')
        self.log_file_size = self.app.config.get("LOG_FILE_SIZE", 300*1024*1024)
        self.setup()

    def setup(self)-> None:
        """
        把日志功能安装到flask项目中
        :return:
        """
        # from logging.handlers import TimedRotatingFileHandler 按时间片分割日志
        handler: TimedRotatingFileHandler = TimedRotatingFileHandler(
            filename=self.log_file,  # 日志存储的文件路径
            when=self.log_rotating_time,  # 每天备份日志的时间，午夜
            backupCount=self.log_backpu_count,  # 备份数量
            encoding=self.log_charseter # 日志文件编码
        )

        # from logging.handlers import RotatingFileHandler      按文件大小分割日志
        # handler: RotatingFileHandler = RotatingFileHandler(
        #     filename=self.log_file,
        #     maxBytes=self.log_file_size,
        #     backupCount=self.log_backpu_count,
        #     encoding=self.log_charseter # 日志文件编码
        # )

        # 设置日志信息的等级
        handler.setLevel(self.log_level)

        # 日志信息的格式
        logging_format: logging.Formatter = logging.Formatter(self.log_format)
        handler.setFormatter(logging_format)

		# 将配置好的日志处理器添加到 Flask 应用的日志系统中
        self.app.logger.addHandler(handler)

application/settings/__init__.py代码：

"""日志配置"""
LOG_FILE: str = "logs/app.log"
LOG_LEVEL: str = "INFO"
LOG_BACKPU_COUNT: int = 31
LOG_FORMAT: str = '%(asctime)s - %(levelname)s - %(filename)s - %(funcName)s - %(lineno)s - %(message)s'

application/settings/dev.py代码：

"""日志配置"""
LOG_FILE: str = "logs/yingming.log"
LOG_LEVEL: str = "DEBUG"
LOG_BACKPU_COUNT: int = 31
LOG_FORMAT: str = '%(asctime)s - %(levelname)s - %(filename)s - %(funcName)s - %(lineno)s - %(message)s'

在 application/__init__.py文件中的init_app 方法中调用日志类进行日志功能的初始化。

from pathlib import Path
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_redis import FlaskRedis
from flask_pymongo import PyMongo

from application.utils.config import Config
from application.utils.logger import Logger

# 实例化配置加载类
config: Config = Config()
# 实例化SQLAlchemy
db: SQLAlchemy = SQLAlchemy()
# 实例化redis
redis_cache: FlaskRedis = FlaskRedis(config_prefix="REDIS")
redis_check: FlaskRedis = FlaskRedis(config_prefix="CHECK")

# mongoDB实例化
mongo: PyMongo = PyMongo()
# 实例化日志配置类
logger: Logger = Logger()


def init_app(config_path: str) -> Flask:
    """用于创建app实例对象并完成初始化过程的工厂函数"""
    # 实例化flask应用对象
    app: Flask = Flask(__name__)
    # 全局路径常量，指向项目根目录
    app.BASE_DIR: Path = Path(__file__).resolve().parent.parent

    # 加载配置
    config.init_app(app, config_path)
    # SQLAlchemy加载配置
    db.init_app(app)
    # redis加载配置
    redis_cache.init_app(app)
    redis_check.init_app(app)
    # # pymongo加载配置
    mongo.init_app(app)

    # 日志加载配置
    logger.init_app(app)

    return app

完成日志初始化配置以后的项目目录结构

yingmingapi/
├── application/
│   ├── __init__.py
│   ├── settings/
│   │   ├── dev.py
│   │   ├── __init__.py
│   │   ├── pro.py
│   └── utils/
│       ├── config.py
│       ├── __init__.py
│       └── logger.py      # 日志相关模块代码
├── logs/                  # 日志文件存储目录
│   └── yingming.log
├── .gitignore
├── command.sh
└── manage.py

经过上面的改造，我们接下来就可以开始创建蓝图了。

提交代码

git add .
git commit -m "fix: add log"
git push

蓝图初始化

在application下创建apps目录，apps以后专门用于保存项目的每一个蓝图应用，并在apps创建home蓝图目录,并在home/__init__.py文件中创建蓝图对象，还有蓝图目录下若干文件，例如视图、模型、路由、序列化器等文件。

.
└── application
    ├── __init__.py
    └── apps/
        └─ home/
           ├─ __init__.py
           ├─ admin.py        # admin站点配置文件
           ├─ serializers.py  # 序列化文件  marshmallow
           ├─ models.py       # mysql模型
           ├─ documents.py     # mongoDB模型
           ├─ tasks.py        # celery的异步任务
           ├─ ws.py           # websocket 的视图文件[websocket接口视图]
           ├─ api.py          # api 的视图文件[api接口视图]
           ├─ views.py        # http web视图接口   [普通视图]
           ├─ urls.py         # 子路由文件
           └─ tests.py        # 单元测试的测试用例

我们可以通过自定义终端命令, 创建一个自动生成蓝图目录的命令.application/utils/commands.py,代码:

import click, os
from typing import Optional
from flask import Flask

class Command(object):
    """Flask终端命令管理类"""
    def __init__(self, app: Optional[Flask] = None):
        if app:
            self.init_app(app)

    def init_app(self, app: Flask):
        self.app: Flask = app
        self.setup()

    def setup(self):
        """初始化终端命令"""
        self.blueprint()  # 自动创建蓝图目录和文件

    def blueprint(self):
        """蓝图目录生成命令"""
        @self.app.cli.command("blue")  # 指定终端命令的调用名称
        @click.option("--name", default="home", help="蓝图目录名称", type=str)
        def command(name: str):
            # 生成蓝图名称对象的目录
            os.mkdir(name)
            open("%s/__init__.py" % name, "w")
            open("%s/admin.py" % name, "w")  # adnmin后台站点配置文件
            open("%s/ws.py" % name, "w")  # websocket的视图文件
            open("%s/api.py" % name, "w")  # api接口的视图文件
            open("%s/views.py" % name, "w")  # 普通视图文件
            open("%s/models.py" % name, "w")
            open("%s/documents.py" % name, "w")
            open("%s/urls.py" % name, "w")  # 视图路由
            open("%s/test.py" % name, "w")
            open("%s/tasks.py" % name, "w")
            open("%s/serializers.py" % name, "w")  # 序列化器文件

            print("BluePrint[%s] created done...." % name)

上面的命令就可以帮我们完成项目中生成蓝图的功能,就下来我们就可以直接把命令注册到manage对象中就可以使用了.

在项目入口文件application/__init__.py中, 初始化终端命令即可。

# 先写 import.. 后写 from .. import ...
# 先解析器内置，后框架官方，然后第三方模块，接着是自己封装的本地模块。

from pathlib import Path

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_redis import FlaskRedis
from flask_pymongo import PyMongo

from application.utils.config import init_config
from application.utils.logger import Log
from application.utils.commands import Command

"""加载组件[单例模式]"""
# SQLAlchemy实例化
db: SQLAlchemy = SQLAlchemy()

# redis实例化
redis_cache: FlaskRedis = FlaskRedis(config_prefix="REDIS")
redis_check: FlaskRedis = FlaskRedis(config_prefix="CHECK")

# mongoDB实例化
mongo: PyMongo = PyMongo()

# 日志实例化
logger:Log = Log()

# 终端命令管理类实例化
command: Command = Command()


def init_app(config_path: str) -> Flask:
    """用于创建app实例对象并完成初始化过程的工厂函数"""

    app: Flask = Flask(__name__)

    # 项目根目录
    app.BASE_DIR = Path(__file__).resolve().parent.parent

    # 初始化项目配置
    init_config(app=app, config_path=config_path)

    # SQLAlchemy加载配置
    db.init_app(app)

    # redis加载配置
    redis_cache.init_app(app)
    redis_check.init_app(app)

    # pymongo加载配置
    mongo.init_app(app)

    # 日志加载配置
    logger.init_app(app)

    # 终端命令管理类加载配置
    command.init_app(app)

    return app

接下来就可以在终端下,通过命令生成蓝图目录了，在项目根目录下创建生成蓝图的shell命令，command.sh：

#!/usr/bin/env bash
export FLASK_APP="/home/moluo/Desktop/yingmingapi/manage.py"
export FLASK_DEBUG=True

if [ $1 ]; then
  if [ $1 == "run" ]; then
    flask run --host=0.0.0.0 --port=5000
  elif [ $1 == "blue" ]; then
    cd application/apps
    flask $1 --name=$2
  else
    flask $1
  fi
fi

终端下调用：

./command.sh blue home

效果:

有了蓝图以后,接下来我们就可以把视图代码,模型代码,路由代码等存储到蓝图目录下了,但是我们需要把蓝图注册到app应用对象下，想想以后，是不是会出现很多的蓝图？

所以我们能不能像django项目中的配置项INSTALLED_APPS里面填写完蓝图路径就可以自动注册蓝图？

可以的，我们也可以封装注册蓝图的功能到一个函数中，让项目自动识别并注册蓝图，并自动加载蓝图下的所有内容。

我们需要在配置文件中声明一个蓝图注册列表（例如：INSTALL_BLUEPRINTS）, 在蓝图自动注册的函数中只注册列表中填写的蓝图。

application/settings/__init__.py，代码：

from typing import List

"""蓝图列表"""
INSTALL_BLUEPRINTS: List = [

]

application/settings/dev.py,代码:

from typing import List


"""蓝图列表"""
INSTALL_BLUEPRINTS: List = [
    "application.apps.home",
]

自动注册蓝图

在application/utils/blueprint.py模块中声明一个auto_resgister_blueprint函数,函数中针对注册到项目配置文件中的INSTALL_BLUEPRINTS蓝图列表选项的蓝图内容实现自动注册到app应用对象里面。

application/utils/blueprint.py,代码:

from typing import List, Optional
from flask import Flask, Blueprint


class AutoBluePrint(object):
    def __init__(self, app: Optional[Flask] = None):
        if app:
            self.init_app(app)

    def init_app(self, app: Flask):
        """
        自动注册蓝图
        :param app:
        :return:
        """
        # 从配置文件中读取需要注册到项目中的蓝图路径信息
        blueprint_path_list: List = app.config.get("INSTALL_BLUEPRINTS", [])
        # 遍历蓝图路径列表，对每一个蓝图进行初始化
        for blueprint_path in blueprint_path_list:
            # 获取蓝图路径中最后一段的包名作为蓝图的名称
            blueprint_name: str = blueprint_path.split(".")[-1]
            # 给当前蓝图目录创建一个蓝图对象
            blueprint: Blueprint = Blueprint(blueprint_name, blueprint_path)
            # 把蓝图对象注册到app实例对象
            # todo url_prefix 是地址前缀，将来我们将来实现一个总路由来声明它
            app.register_blueprint(blueprint, url_prefix="")

        print(app.blueprints)

项目入口文件application/__init__.py中, 实例化AutoBluePrint类, 自动生成并注册蓝图对象，application/__init__.py,代码:

from pathlib import Path
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_redis import FlaskRedis
from flask_pymongo import PyMongo

from application.utils.config import Config
from application.utils.logger import Logger
from application.utils.commands import Command
from application.utils.blueprint import AutoBluePrint


# 实例化配置加载类
config: Config = Config()
# 实例化SQLAlchemy
db: SQLAlchemy = SQLAlchemy()
# 实例化redis
redis_cache: FlaskRedis = FlaskRedis(config_prefix="REDIS")
redis_check: FlaskRedis = FlaskRedis(config_prefix="CHECK")

# mongoDB实例化
mongo: PyMongo = PyMongo()
# 实例化日志配置类
logger: Logger = Logger()

# 实例化终端命令管理类
command: Command = Command()

# 实例化自动化蓝图类
blueprint: AutoBluePrint = AutoBluePrint()

def init_app(config_path: str) -> Flask:
    """用于创建app实例对象并完成初始化过程的工厂函数"""
    # 实例化flask应用对象
    app: Flask = Flask(__name__)
    # 全局路径常量，指向项目根目录
    app.BASE_DIR: Path = Path(__file__).resolve().parent.parent

    # 加载配置
    config.init_app(app, config_path)
    # SQLAlchemy加载配置
    db.init_app(app)
    # redis加载配置
    redis_cache.init_app(app)
    redis_check.init_app(app)
    # # pymongo加载配置
    mongo.init_app(app)

    # 日志加载配置
    logger.init_app(app)

    # 终端命令管理类加载配置
    command.init_app(app)

    # 自动化蓝图类加载配置
    blueprint.init_app(app)

    return app

注册了蓝图对象以后,蓝图下面的视图方法和视图对应的路由关系也要进行注册!

注册蓝图的路由和视图

在蓝图home下面的urls.py文件中,通过path方法把url地址和视图方法、http请求、处理成字典，然后把字典作为成员添加到路由列表的固定变量urlpatterns列表中.

application/utils/blueprint.py,声明path函数,代码：

from types import ModuleType
from typing import List, Optional, Callable
from flask import Flask, Blueprint
from importlib import import_module


class AutoBluePrint(object):
    def __init__(self, app: Optional[Flask] = None):
        if app:
            self.init_app(app)

    def init_app(self, app: Flask):
        """
        自动注册蓝图
        :param app:
        :return:
        """
        # 从配置文件中读取需要注册到项目中的蓝图路径信息
        blueprint_path_list: List = app.config.get("INSTALL_BLUEPRINTS", [])
        # 遍历蓝图路径列表，对每一个蓝图进行初始化
        for blueprint_path in blueprint_path_list:
            # 获取蓝图路径中最后一段的包名作为蓝图的名称
            blueprint_name: str = blueprint_path.split(".")[-1]
            # 给当前蓝图目录创建一个蓝图对象
            blueprint: Blueprint = Blueprint(blueprint_name, blueprint_path)

            # 导入子路由关系，blueprint_url_path就是当前蓝图下的urls模块的导包路径
            blueprint_url_path: str = blueprint_path + ".urls"
            urls_module: ModuleType = import_module(blueprint_url_path)
            urlpatterns: List = []
            try:
                urlpatterns = urls_module.urlpatterns
            except Exception:
                pass

            # 在循环中，把urlpatterns的每一个路由信息添加注册到蓝图对象里面
            for url in urlpatterns:
                blueprint.add_url_rule(**url)

            # 最后把蓝图对象注册到app实例对象
            # todo url_prefix 是地址前缀，将来我们将来实现一个总路由来声明它
            app.register_blueprint(blueprint, url_prefix="")

def path(rule: str, view_func: Callable, **kwargs):
    """绑定url地址和视图的映射关系"""
    return {"rule": rule, "view_func": view_func, **kwargs}

在application.__init__入口文件中，导入path路由映射函数，将来方便再其他地方调用该函数，application.__init__，文件添加代码如下：

from application.utils.blueprint import AutoBluePrint, path

在home蓝图下的urls.py中,使用path函数绑定视图和路由的映射关系并添加urlpatterns中,home.urls.py,代码：

    from typing import List
    from application import path
    from . import views

    urlpatterns: List = [
        # path的作用就是把传递的参数转换成对应的字典结构：{"rule": "/home", "view_func": views.index, "methods": ["GET"]},
        path("/test", views.test, methods=["GET", "POST"]),
        path("/index", views.index),
    ]

为了方便测试，先到当前home蓝图下视图文件views.py中，添加测试的视图，代码：

def index():
    return "home.index"

def test():
    return "home.test"

完成上面步骤，将来即便有新的蓝图也可以参考上面的写法，把当前蓝图中所有的路由全部集中在urlpatterns中，接着下来我们就可以在application.utils.blueprint.AuthBluePrint自动化蓝图类中, 生成蓝图对象以后添加自动加载并注册蓝图的路由和视图了。

此时,运行项目，就可以通过url地址访问蓝图下的视图方法了。

GET http://127.0.0.1:5000/index

POST http://127.0.0.1:5000/test

提交代码版本

git add .
git commit -m "fix: auto register blueprint"
git push

上面蓝图注册到app实例对象时, 没有设置url_prefix路由前缀, 接下来我们可以单独设置一个总路由application/urls.py,进行路由前缀的设置.

在项目默认配置文件中,application/settings/__init__.py,新增总路由的配置项URL_PATH

"""总路由"""
URL_ROOT_PATH = "application.urls"

接下来,在application.utils.blueprint文件中调整path函数中接受接受参数类型，把路由前缀和蓝图名称进行绑定映射，application.utils.blueprint，代码：

def path(rule: str, name: Union[Callable, str], **kwargs) -> Dict:
    """绑定url地址和视图的映射关系"""
    if isinstance(name, Callable):
        return {"rule": rule, "view_func": name, **kwargs}
    elif isinstance(name, str):
        return {"url_prefix": rule, "blueprint_url_subffix": name, **kwargs}
    else:
        return {}

创建总路由文件并注册蓝图和路由前缀的关系, application/urls.py,代码:

from typing import List
from application import path


urlpatterns: List = [
    path("/home", "home.urls"),
]

接下来，就可以在注册蓝图时，把路由前缀随着蓝图对象一起注册到app实例对象中。application.utils.blueprint，代码：

from types import ModuleType
from typing import List, Optional, Callable, Union, Dict
from flask import Flask, Blueprint
from importlib import import_module


class AutoBluePrint(object):
    def __init__(self, app: Optional[Flask] = None):
        if app:
            self.init_app(app)

    def init_app(self, app: Flask):
        """
        自动注册蓝图
        :param app:
        :return:
        """
        # 从配置文件中读取需要注册到项目中的蓝图路径信息
        blueprint_path_list: List = app.config.get("INSTALL_BLUEPRINTS", [])
        # 从配置文件中读取总路由模块
        url_root_path: str = app.config.get("URL_ROOT_PATH", "application.urls")
        # 总路由模块
        url_root_module: ModuleType = import_module(url_root_path)
        # 总路由列表
        if not hasattr(url_root_module, "urlpatterns"):
            message: str = "总路由文件 URL_ROOT_PATH，没有路由列表！请在总路由文件中设置 urlpatterns 路由列表"
            app.logger.error(message)
            raise Exception(message)

        root_urlpatterns: List = url_root_module.urlpatterns

        # 遍历蓝图路径列表，对每一个蓝图进行初始化
        for blueprint_path in blueprint_path_list:
            # 获取蓝图路径中最后一段的包名作为蓝图的名称
            blueprint_name: str = blueprint_path.split(".")[-1]
            # 给当前蓝图目录创建一个蓝图对象
            blueprint: Blueprint = Blueprint(blueprint_name, blueprint_path)

            # # 导入子路由关系，blueprint_url_path就是当前蓝图下的urls模块的导包路径
            # blueprint_url_path: str = blueprint_path + ".urls"
            # urls_module: ModuleType = import_module(blueprint_url_path)
            # urlpatterns: List = []
            # try:
            #     urlpatterns = urls_module.urlpatterns
            # except Exception:
            #     pass
            #
            # # 在循环中，把urlpatterns的每一个路由信息添加注册到蓝图对象里面
            # for url in urlpatterns:
            #     blueprint.add_url_rule(**url)

            # 蓝图路由的前缀
            url_prefix: str = ""

            # 蓝图下的子路由列表
            urlpatterns: List = []

            # 获取蓝图的父级目录，目的是为了拼接总路由中所有蓝图下的urls子路由文件的路径
            blueprint_father_path: str = ".".join(blueprint_path.split(".")[:-1])

            # 循环总路由列表
            for item in root_urlpatterns:
                # 判断当前蓝图是否有注册到总路由中提供对外访问，如果没有把蓝图注册到总路由中，则无法被外界访问。
                if blueprint_name in item["blueprint_url_subffix"]:
                    # 导入当前蓝图下的子路由模块
                    urls_module: ModuleType = import_module(f"{blueprint_father_path}.{item['blueprint_url_subffix']}")
                    if hasattr(urls_module, "urlpatterns"):
                        # 获取子路由文件中的路由列表
                        urlpatterns: List = urls_module.urlpatterns
                    # 提取蓝图路由的前缀
                    url_prefix = item["url_prefix"]
                    # 把urlpatterns的每一个路由信息添加注册到蓝图对象里面
                    for url in urlpatterns:
                        blueprint.add_url_rule(**url)
                    break

            # 最后把蓝图对象注册到app实例对象
            # url_prefix 是地址前缀，将来我们将来实现一个总路由来声明它
            app.register_blueprint(blueprint, url_prefix=url_prefix)


def path(rule: str, name: Union[Callable, str], **kwargs) -> Dict:
    """绑定url地址和视图的映射关系"""
    if isinstance(name, Callable):
        return {"rule": rule, "view_func": name, **kwargs}
    elif isinstance(name, str):
        return {"url_prefix": rule, "blueprint_url_subffix": name, **kwargs}
    else:
        return {}

提交版本

git add .
git commit -m "fix: add blueprint's url_perfix"
git push

蓝图模型的自动注册

在蓝图目录home下的models.py中声明模型,例如：

from application import db

class User(db.Model):
    __tablename__ = "ym_user"
    id = db.Column(db.Integer, primary_key=True, comment="主键ID")
    name = db.Column(db.String(255), unique=True, comment="账户名")
    password = db.Column(db.String(255), comment="登录密码")
    ip_address = db.Column(db.String(255), index=True, comment="登录IP")

    def __repr__(self):
        return self.name

完成上面模型声明以后，我们在入口文件application.__init__中通过 db的 create_all() 创建数据表。

application.__init__，代码：

from pathlib import Path
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_redis import FlaskRedis
from flask_pymongo import PyMongo

from application.utils.config import Config
from application.utils.logger import Logger
from application.utils.commands import Command
from application.utils.blueprint import AutoBluePrint, path


# 实例化配置加载类
config: Config = Config()
# 实例化SQLAlchemy
db: SQLAlchemy = SQLAlchemy()
# 实例化redis
redis_cache: FlaskRedis = FlaskRedis(config_prefix="REDIS")
redis_check: FlaskRedis = FlaskRedis(config_prefix="CHECK")

# mongoDB实例化
mongo: PyMongo = PyMongo()
# 实例化日志配置类
logger: Logger = Logger()

# 实例化终端命令管理类
command: Command = Command()

# 实例化自动化蓝图类
blueprint: AutoBluePrint = AutoBluePrint()

def init_app(config_path: str) -> Flask:
    """用于创建app实例对象并完成初始化过程的工厂函数"""
    # 实例化flask应用对象
    app: Flask = Flask(__name__)
    # 全局路径常量，指向项目根目录
    app.BASE_DIR: Path = Path(__file__).resolve().parent.parent

    # 加载配置
    config.init_app(app, config_path)
    # SQLAlchemy加载配置
    db.init_app(app)
    # redis加载配置
    redis_cache.init_app(app)
    redis_check.init_app(app)
    # # pymongo加载配置
    mongo.init_app(app)

    # 日志加载配置
    logger.init_app(app)

    # 终端命令管理类加载配置
    command.init_app(app)

    # 自动化蓝图类加载配置
    blueprint.init_app(app)

    # db创建数据表
    with app.app_context():
        db.create_all()

    return app

项目重启，发现flask根本不知道我们已经声明上面的模型，所以根本没有创建数据表，这个原因是，我们模型写在了蓝图目录下，我们需要把模型导入到要么和蓝图对象一起，要么和app实例对象一起。因为模型是被蓝图保存的，所以我们可以在蓝图对象创建中，导入模型，让flask识别到。

application/utils/blueprint.py,代码:

from types import ModuleType
from typing import List, Optional, Callable, Union, Dict
from flask import Flask, Blueprint
from importlib import import_module


class AutoBluePrint(object):
    def __init__(self, app: Optional[Flask] = None):
        if app:
            self.init_app(app)

    def init_app(self, app: Flask):
        """
        自动注册蓝图
        :param app:
        :return:
        """
        # 从配置文件中读取需要注册到项目中的蓝图路径信息
        blueprint_path_list: List = app.config.get("INSTALL_BLUEPRINTS", [])
        # 从配置文件中读取总路由模块
        url_root_path: str = app.config.get("URL_ROOT_PATH", "application.urls")
        # 总路由模块
        url_root_module: ModuleType = import_module(url_root_path)
        # 总路由列表
        if not hasattr(url_root_module, "urlpatterns"):
            message: str = "总路由文件 URL_ROOT_PATH，没有路由列表！请在总路由文件中设置 urlpatterns 路由列表"
            app.logger.error(message)
            raise Exception(message)

        root_urlpatterns: List = url_root_module.urlpatterns

        # 遍历蓝图路径列表，对每一个蓝图进行初始化
        for blueprint_path in blueprint_path_list:
            # 获取蓝图路径中最后一段的包名作为蓝图的名称
            blueprint_name: str = blueprint_path.split(".")[-1]
            # 给当前蓝图目录创建一个蓝图对象
            blueprint: Blueprint = Blueprint(blueprint_name, blueprint_path)

            # # 导入子路由关系，blueprint_url_path就是当前蓝图下的urls模块的导包路径
            # blueprint_url_path: str = blueprint_path + ".urls"
            # urls_module: ModuleType = import_module(blueprint_url_path)
            # urlpatterns: List = []
            # try:
            #     urlpatterns = urls_module.urlpatterns
            # except Exception:
            #     pass
            #
            # # 在循环中，把urlpatterns的每一个路由信息添加注册到蓝图对象里面
            # for url in urlpatterns:
            #     blueprint.add_url_rule(**url)

            # 蓝图路由的前缀
            url_prefix: str = ""

            # 蓝图下的子路由列表
            urlpatterns: List = []

            # 获取蓝图的父级目录，目的是为了拼接总路由中所有蓝图下的urls子路由文件的路径
            blueprint_father_path: str = ".".join(blueprint_path.split(".")[:-1])

            # 循环总路由列表
            for item in root_urlpatterns:
                # 判断当前蓝图是否有注册到总路由中提供对外访问，如果没有把蓝图注册到总路由中，则无法被外界访问。
                if blueprint_name in item["blueprint_url_subffix"]:
                    # 导入当前蓝图下的子路由模块
                    urls_module: ModuleType = import_module(f"{blueprint_father_path}.{item['blueprint_url_subffix']}")
                    if hasattr(urls_module, "urlpatterns"):
                        # 获取子路由文件中的路由列表
                        urlpatterns: List = urls_module.urlpatterns
                    # 提取蓝图路由的前缀
                    url_prefix = item["url_prefix"]
                    # 把urlpatterns的每一个路由信息添加注册到蓝图对象里面
                    for url in urlpatterns:
                        blueprint.add_url_rule(**url)
                    break

            try:
                # 让蓝图自动发现模型模块
                import_module(f"{blueprint_path}.models")
            except ModuleNotFoundError:
                pass

            # 最后把蓝图对象注册到app实例对象
            # url_prefix 是地址前缀，将来我们将来实现一个总路由来声明它
            app.register_blueprint(blueprint, url_prefix=url_prefix)


def path(rule: str, name: Union[Callable, str], **kwargs) -> Dict:
    """绑定url地址和视图的映射关系"""
    if isinstance(name, Callable):
        return {"rule": rule, "view_func": name, **kwargs}
    elif isinstance(name, str):
        return {"url_prefix": rule, "blueprint_url_subffix": name, **kwargs}
    else:
        return {}

提交版本

git add .
git commit -m "fix: auto find models"
git push

项目能自动加载总路由也能加载蓝图下的子路由和视图模型以后的项目目录结构，如下:

yingmingapi/
├── logs/                    # 日志目录
├── applicaton/             # 项目主要逻辑代码保存目录
 |   ├── settings/           # 项目配置存储目录
│   │   ├ __init__.py       # 项目默认初始化配置文件
│   │   ├ dev.py            # 开发阶段的配置文件
│   │   └ pro.py            # 生产阶段的配置文件
│   ├── __init__.py         # 项目初始化[全局入口文件]
 |   ├── utils/              # 项目工具类库目录
│   │   ├ blueprint.py      # 蓝图注册相关的函数或工具类
│   │   ├ commands.py       # 自定义命令和加载命令的相关函数
│   │   ├ config.py         # 项目配置加载的辅助函数
│   │   └ logger.py         # 日志模块
│   ├── apps/               # 保存项目中所有蓝图的存储目录
│   │   ├── home            # 蓝图目录【这里是举例而已】
│   │   │   ├── __init__.py # 蓝图的初始化文件
│   │   │   ├── urls.py     # 蓝图的子路由文件
│   │   │   ├── models.py   # 蓝图的模型文件
│   │   │   └── views.py    # 蓝图的视图文件
│   │   ├── __init__.py
│   └── urls.py             # 总路由
├── manage.py               # 项目的终端管理脚本文件
└── command.sh              # 启动命令

若有错误与不足请指出，关注DPT一起进步吧！！！