基于Flask框架的API对接项目实战演示源码

flask-api-demo Python Python, HTML 共45个文件 py: 36 yml: 2 gitignore: 1 json: 1 md: 1 db: 1 ico: 1 html: 1 txt: 1 整合一个基于flask的api对接项目demo。 <center>flask-api-demo</center> 一个基于flask架构的 接口/数据对接 小型项目开发结构模板。 封装了一些必要插件如 日志记录，定时任务，数据库操作，自定义错误，在线接口文档等 项目结构介绍 ├───app 程序主体 │ ├───module 项目代码位置 │ │ └───项目*自定义的项目* │ ├───plugin 通用功能插件 │ │ └───__init__.py 定义所有插件对外接口方法 | | └───com_method.py 一些可能用得上的通用方法记录在这 │ │ └───db.py 数据库操作封装 │ │ └───exception.py 自定义异常封装 │ │ └───falsk_docs.py 在线接口文档自定义封装 │ │ └───logger.py 日志封装 │ │ └───myrandom.py 随机数封装，包括汉字，英文大小写随机 │ │ └───mytime.py 时间格式转化封装 │ │ └───resp.py 通用接口返回封装 │ │ └───scheduler.py 定时任务封装 │ ├───base.py 副主体文件，全局的一些定义和配置 │ ├───config_apidoc.yml 在线文档的配置文件 │ ├───config.py 配置文件读取器 │ ├───config.yml 配置文件 │ ├───main.py 主体文件，为了精简，将部分主体代码转移到base.py │ ├───nb_log_config.py 日志配置文件，nb_log包自带，不建议修改 ├───.gitignore git 提交忽略文件 ├───data.db 本地数据库文件 ├───requirements.txt 项目运行环境依赖包 ├───README.md 项目说明文件 运行 python ./app/main.py 打包 pip install pyinstaller # 注意flask_docs的路径，不同电脑会不同 # --add-data表示打包资源文件，静态文件需要额外打包 pyinstaller --add-data "C:\Users\chens\AppData\Local\Programs\Python\Python310\Lib\site-packages\flask_docs;flask_docs" --add-data "./app/view;view" -F ./app/main.py [-n saftopclientAPI] 可以将程序打包成exe文件，注意，静态文件不会打包进去，需要手动将config.yml,config_apidoc.yml,data.db放到exe文件同目录进行运行 使用介绍 新增对接一个项目 在module路径下，新建一个项目文件夹，里面可以任意实现。 推荐在自定义文件夹下面新建一个__init__.py,引入项目的运行主体方法。如果是接口，引入蓝图。 main.py里面import对应的项目，并调用方法主体/引用蓝图即可。 例： ## file:module/demo/demo_handle.py from flask import Blueprint demo = Blueprint('demo',__name__,url_prefix='/demo') # 接口示例 @demo.route('/test') def test(): return 'success' # 方法示例 def demo_main(): sync_data() # 省略实现，假设是实现一个数据同步 ----------------------------------------------------------------- ## file：module/demo/__init__.py from module.demo.demo_handle import demo_main,demo demo_main # 方法主体 demo # 接口蓝图 ----------------------------------------------------------------- ## file: main.py ## 修改如下： # coding: utf-8 """ 运行文件，尽量简洁 """ from base import app, register_blueprints from module.demo import demo_main,demo # 这里引用项目 @app.route('/') def home(): return 'welcome to saftop api module home page' if __name__ == '__main__': demo_main() # 执行方法主体 app.config['BluePrints'] = [demo] # 所有蓝图放到这里面来,下面的方法会读取配置并注册蓝图 app.config['API_DOC_MEMBER'] = [demo] # 所有希望生成接口文档得蓝图放到这里来 register_blueprints() app.run(host='0.0.0.0', port=5051) 插件介绍 所有插件对外开放的接口方法，定义到plugin.__init__.py里面,引用时使用 from plugin imort xxx,不要直接引用文件 数据库操作 目前封装了sqlserver、mysql、sqlite以及redis 使用了dbutils的线程池，默认5：min--10：max个数据库的短连接，使用次数为100。可以通过传参自定义。 线程池根据数据库连接生成，不同的数据库连接在第一次初始化时都会建立线程池，多次初始化同一个数据库连接不会影响 以sqlserver为例介绍功能 """ # 初始化连接对象 # 线程池变量以host+port+database+user+password+charset为key，数据库连接为value # 同一个数据库连接可以随意实例化，不会影响连接和效率 """ class MSsqlClient(host, port, database, user, password, charset='utf8', mincached=5, maxcached=10, maxconnections=100, blocking=True, maxusage=100) """ 通过脚本查询结果，返回一个结果元组 """ def MSsqlClient().select(sql[,param])->tuple """ 通过脚本查询结果，将结果和字段名称组合成字典返回 """ def MSsqlClient().select_to_dict(sql[,param])->list(dict) """ 执行增删改语句，返回影响行数 """ def MSsqlClient().execute(sql[,param])->int """ 开启事务 """ def MSsqlClient().begin()->none """ 结束事务，option为commin时提交，其他时回滚 """ def MSsqlClient().end(option='commit')->int """ 执行语句，不提交，配合事务使用 """ def MSsqlClient().execute_without_commit(sql[,param])->int 日志操作 日志使用的是国人开发的nb_log包，性能非常好，功能全面，就是使用起来会比loguru麻烦一些。 最方便的是可以实例化多个日志对象，将日志写入到不同的文件，可以根据项目,日志级别等来分日志。 封装如下： 隔离了一些配置，简化了使用。 定义了默认日志对象InnerLogger.log,写入文件为./log/base.log,日志级别在config里面定义 定义了flask内部api调用日志，写入文件为./log/invoke_api.log """ 使用示例 """ from plugin import log, get_log log.info('xxx') # 使用默认日志对象记录日志 demo_log = get_log('demo','./log/demo.log',10) demo_log.error('xxx') # 使用自定义日志对象记录日志 """ 日志接口说明 """ """ # 默认日志对象，各插件使用的日志对象，项目中可以直接使用这个对象记录日志，也可以自定义日志对象 """ class InnerLogger.log """ # 生成一个日志对象，默认级别为error @param :module: 日志名称，根据不同的名称生成不同的日志对象，名称相同则返回的日志对象相同 :path: 完整的日志路径和名称 :level: 默认error """ def get_log(module, path, level) 定时任务 使用的aspschduler包，BackgroundScheduler类。 支持普通定时任务和持久化定时任务。 独立定时任务 使用@SchedulerManager.Scheduler(Trigger),推荐 使用SchedulerManager.add_job(func,Trigger) """ 使用示例 """ from plugin import SchedulerManager, Trigger import time @SchedulerManager.Scheduler(Trigger.interval(minutes=10)) # Trigger是自定义类，方便设定任务参数 def a(): print(time.time()) a() # 调用方法时会将方法加到定时器里面去，并默认当前执行一遍 """ 接口定义 """ """ 作业任务管理，管理所有的定时任务 """ class SchedulerManager(): """ 任务配置 """ JOB_DEFAULTS_CONFIG = { 'coalesce': True, # 过时任务是否忽略 'replace_existing': True, # 如果存在相同id任