Flask笔记-任务框架搭建

这个是个很好的结构，如下：

先贴下代码：

base_setting.py

DEBUG = True

local_setting.py

local_setting.py

production_setting.py

from config.base_setting import *
 
DEBUG = False

test.py

 
class JobTask():
    def __init__(self):
        pass
 
    def run(self, params):
        print("Job测试打印")
        print(params)
        pass

launcher.py

from flask_script import Command
import sys, argparse, traceback, importlib
 
'''
Job统一入口文件
python manager.py runjob -m Test
python manager.py runjob -m test/index
第一个jobs/tasks/Test.py
第二个jobs/tasks/test/index.py
'''
 
class runJob(Command):
 
    capture_all_args = True     #flask_script有一个未记录的capture_all_flags flag，它将把剩余的参数传递给Command.run方法
    def run(self, *args, **kwargs):
        args = sys.argv[2:]
        print(args)
        parser = argparse.ArgumentParser(add_help=True)
        parser.add_argument("-m", "--name", dest="name", metavar="name", help="指定job名", required=True)
        parser.add_argument("-a", "--act", dest="act", metavar="act", help="Job动作", required=False)
        parser.add_argument("-p", "--param", dest="param", nargs="*", metavar="param", help="业务参数", required=False)
 
        params = parser.parse_args(args)
        params_dict = params.__dict__
        print(params)
        print(params_dict)
 
        if "name" not in params_dict or not params_dict['name']:
            return self.tips()
 
        try:
            '''
            类似java的反射，导入包
            from jobs.xxx.xxx import JobTask
            '''
            module_name = params_dict['name'].replace("/", ".")
            import_string = "jobs.tasks.%s" % (module_name)
            target = importlib.import_module(import_string)
            exit(target.JobTask().run(params_dict))
        except Exception as e:
            traceback.print_exc()
        pass
 
    def tips(self):
        tip_msg = '''
        请正确的调度Job
        python manager.py runjob -m test ( jobs/tasks/test.py )
        python manager.py runjob -m test/index ( jobs/tasks/test/index.py )
        '''
        print( tip_msg )
        return
 
    pass

application.py

from flask import Flask
from flask_script import Manager
import os
 
app = Flask(__name__)
 
manager = Manager(app)
 
app.config.from_pyfile("config/base_setting.py")
if "ops_config" in os.environ:
    app.config.from_pyfile("config/%s_setting.py" % (os.environ['ops_config']))
    pass

manager.py

from flask_script import Server, Command
from application import app, manager
from jobs.launcher import runJob
 
manager.add_command("runjob", runJob)
 
def main():
    manager.run()
    pass
 
if __name__ == "__main__":
    try:
        import sys
        sys.exit(main())
        pass
    except Exception as e:
        import traceback
        traceback.print_exc()
        pass
    pass

requirement.txt

flask
flask-script
apscheduler
Flask-APScheduler
requests
BeautifulSoup4

程序是这样运行的，如下：

requirement.txt是库文件，在部署开发环境时，这样使用：

pip install -r requirement.txt

manager.py是主要的管理类。程序的入口。

application.py为核心程序，管理Flask中的app，和flask_script的Manager，并且加载config下的配置文件。

这里开发环境会加载：

base_setting.py + local_setting.py

生产环境会加载：

base_setting.py + production_setting.py

 

launcher.py为任务的入口文件，主要是管理参数方面的问题。

在jobs文件夹下的tasks为具体的任务文件。