基于flask的后端敏捷开发框架详细文档+全部资料+高分项目.zip

## FlaskFD 基于flask的后端敏捷开发框架，围绕多个项目的业务代码进行整合调整，模块抽象，易于理解和助力快速开发。 * flask 引入 gevent、cors，提高并发和兼容跨域 * 精简整合路由配置，还原经典controller、service、dao后台架构分层 * 引入多种业务装饰器，大幅减少前后端数据格式化、业务异常捕获、mysql事务查询等高频次耦合性代码 ## run 开发 ``` # 拉取项目代码 git clone [email protected]:17621192638/flaskfd.git # 安装依赖 pip3 install -r /home/flaskfd/requirements.txt # 运行 python3 /home/flaskfd/web_api.py ``` ## FileTree 目录结构 ``` ┌── FlaskFD ├── controller // 控制器层 ├── service // 业务逻辑层 ├── dao // 数据库层 │ ├── mysql // mysql数据库 │ ├── ├── localhost1 // 数据库1 │ ├── ├── localhost2 // 数据库2 │ ├── ├── conf.ini // mysql数据库配置文件 ├── utils // 项目工具类，基本通用方法汇总 │ ├── mysql │ ├── nlp │ ├── code_util.py // 自定义返回状态码 │ ├── common_util.py // 公共基础方法类 │ ├── decorator_util.py // 通用修饰器 │ ├── faker_util.py │ ├── logger_util.py // 日志工具类 │ ├── response_util.py // 通用response解释器 ├── .gitignore ├── requirements.txt // 项目依赖 ├── web_api.py // api路由配置和项目启动服务 └── ... ``` ## RestFul Api 接口路由配置 ``` web_api.py ... ... import controller.user_controller as user_controller api.add_resource(user_controller.do_login, '/api/login', endpoint="user_controller.do_login") # import controller.domain_controller as domain_controller # api.add_resource(domain_controller.create, '/api/domain/create', endpoint="domain_controller.create") ## import controller.project_controller as project_controller # api.add_resource(project_controller.create, '/api/project/create', endpoint="project_controller.create") ... ... ``` 在web_api.py 项目启动文件中进行api接口路由的配置，引入不同的controller，每个controller方法对应多个api方法，自定义api的地址。 ## Controller配置 ``` user_controller.py ... ... from flask_restful import Resource, request from  service.user_service import  user_service import utils.response_util as response_util service =user_service() class do_login(Resource):    @response_util.response_filter_v2    def post(self):        text_json = request.get_json()        return service.do_login(text_json) ... ... ``` flask支持多种路由配置方式，建议统一风格通过class的形式进行单路由配置。每个controller可以对应一个/多个service服务. **@response_util.response_filter_v2 修饰器**将会对service主体进行异常捕获，异常提示，返回内容格式化，耗时统计等操作 * **内容格式化** 通过对service返回值进行拦截判断与处理，返回给前端统一的接口格式（JSON） ``` { "code": 200, # 返回状态码，接口正常，返回200 "data": res, # 内容主体 "timeout":timeout, # 接口耗时 "message":message # 备注信息 } ``` * **异常捕获** 对service异常代码进行捕获返回 ``` { "code":400, # 返回状态码 "data": None, "timeout":timeout, # 接口耗时 "message":"traceback message" # 代码异常堆栈信息 } ``` * **自定义code类型** 通过对 **utils/code_util.py** 进行添加修改删除自定义状态码，返回给前端特殊code代码和message补充信息（与项目前端协商） ``` { "code":10001, "data": None, "timeout":timeout, "message":"登录失败,账号或者密码错误." } ``` 可以补充在Controller层实现接口权限控制、请求数据统计等功能。 ## Service处理 ``` user_service.py ... ... import dao.mysql.localhost.user_dao as user_dao import utils.common_util as common_util import utils.decorator_util as decorator_util class user_service:    @decorator_util.decorator_none def do_login(self,text_json):        users = user_dao.select(model{"username":text_json["username"],"password":text_json["password"],"status":0})        if not users:return {"code":10001}        else: return users[0] ... ... ``` 具体业务逻辑层，一个service对应一个/多个dao，按具体业务类型和模块进行拆分。 **@decorator_util.decorator_none** 公共拦截器，拦截捕获业务逻辑异常，返回None。 主要应用于局部service业务逻辑处理方法，减少重复 **try exception**代码。通过自定义拦截器 **uitls/decorator_util**，完成对业务逻辑的统一处理。 ## Dao层封装 ### 连接多种数据库 ``` ... ├── dao // 数据库层 │ ├── mysql // mysql数据库 │ ├── elasticsearch // 其它数据库 ... ``` 项目中存在多个数据库的类型处理的与连接，都放到 **dao/** 文件夹下，如mysql、elasticsearch、mongo等 ### 连接多个mysql数据库 ``` ... │ ├── mysql // mysql数据库 │ ├── ├── localhost1 // 数据库1 │ ├── ├── localhost2 // 数据库2 │ ├── ├── conf.ini // 数据库配置文件 ... ``` 多个同数据库类型的库方法按照数据库名称放到**dao/mysql/**文件夹下。 ``` pool.py ... ... key = "localhost" # 创建当前数据库的连接池对象 class service(object):    def __init__(self):        environment = cf.get(key,"environment")        for i in range(3):            print("当前mysql运行环境: {} !!!!".format(environment))            time.sleep(0.1)        self.pool = mysql_util.get_mysql_pool(            host=cf.get(key,"host"),            port=cf.get(key,"port"),            user=cf.get(key,"user"),            password=cf.get(key,"passwd"),            database=cf.get(key,"db"),            charset="utf8mb4"        )        self.conn,self.cursor =mysql_util.get_db_from_pool(pool=self.pool) s = service() run_sql = mysql_util.get_wrapper(s.pool) run_sql_v2 = mysql_util.get_wrapper_v2(s.pool) ... ... from utils.mysql.common_dao import common_dao as common # 给公共方法加上带pool的注释器，无法通过init方法传递注释器 class common_dao(common):    def __init__(self,table_name):        super().__init__(table_name=table_name)    @mysql_util.pymysql_time_deal    @run_sql(fetch_type="all")    def select(self,*args, **kwargs):        return super().select(*args, **kwargs) ... ``` * **pool.py** 每个数据库下面都有一个统一的pool.py文件，通过修改**key**的值，指定加载对应的conf.ini配置文件信息，pool.py在被引用时会初始化并生成一个**mysql连接池对象**。**仅需修改文件中的key值即可**。 * **run_sql** pool.py文件会向外暴露一个run_sql方法，仔细看其实是一个装饰器，被该装饰器装饰的sql语句，会从连接池中获取一个mysql连接对象，去执行sql语句， 并返回执行结果。装饰器封装了连接池**获取连接对象，提交事务，执行事务，按条件执行sql，返回sql结果和异常捕获**相关的逻辑操作代码，灵活运用@run_sql可大幅减少sql相关的重复流程性代码。 * @run_sql(fetch_type="all") 对应返回 cursor.fetchall()方法结果 * @run_sql(fetch_type="one") 对应返回 cursor.fetchone()方法结果 * @run_sql(fetch_typ