這個(gè)是個(gè)很好的結(jié)構(gòu)，如下：

先貼下代碼：

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):passdef run(self, params):print("Job測(cè)試打印")print(params)pass

launcher.py

from flask_script import Command import sys, argparse, traceback, importlib''' Job統(tǒng)一入口文件 python manager.py runjob -m Test python manager.py runjob -m test/index第一個(gè)jobs/tasks/Test.py 第二個(gè)jobs/tasks/test/index.py '''class runJob(Command):capture_all_args = True #flask_script有一個(gè)未記錄的capture_all_flags flag，它將把剩余的參數(shù)傳遞給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動(dòng)作", required=False)parser.add_argument("-p", "--param", dest="param", nargs="*", metavar="param", help="業(yè)務(wù)參數(shù)", 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:'''類(lèi)似java的反射，導(dǎo)入包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()passdef tips(self):tip_msg = '''請(qǐng)正確的調(diào)度Jobpython manager.py runjob -m test ( jobs/tasks/test.py )python manager.py runjob -m test/index ( jobs/tasks/test/index.py )'''print( tip_msg )returnpass

application.py

from flask import Flask from flask_script import Manager import osapp = 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 runJobmanager.add_command("runjob", runJob)def main():manager.run()passif __name__ == "__main__":try:import syssys.exit(main())passexcept Exception as e:import tracebacktraceback.print_exc()passpass

requirement.txt

flask flask-script apscheduler Flask-APScheduler requests BeautifulSoup4

程序是這樣運(yùn)行的，如下：

requirement.txt是庫(kù)文件，在部署開(kāi)發(fā)環(huán)境時(shí)，這樣使用：

pip install -r requirement.txt

manager.py是主要的管理類(lèi)。程序的入口。

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

這里開(kāi)發(fā)環(huán)境會(huì)加載：

base_setting.py + local_setting.py

生產(chǎn)環(huán)境會(huì)加載：

base_setting.py + production_setting.py

?

launcher.py為任務(wù)的入口文件，主要是管理參數(shù)方面的問(wèn)題。

在jobs文件夾下的tasks為具體的任務(wù)文件。?

總結(jié)

以上是生活随笔為你收集整理的Flask笔记-任务框架搭建的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。