order/jobs/launcher.py

# -*- coding: utf-8 -*-
from application import app
import click
import sys
import traceback

'''
python flask runjob -m Test  (  jobs/tasks/Test.py )
python flask runjob -m test/Index (  jobs/tasks/test/Index.py )
* name or flags - 名称或选项字符串列表, e.g. foo or -f, --foo.
* action - 参数如果定义了选项，表示这是一个操作参数，至于调用时做哪种操作由用户输入或者default决定。
* nargs - 应该使用的命令行参数数。.
* const - 某些动作或参数个数的常数值。.
* default - 如果命令行没有对输入这个参数相应的值，则此参数用default给出的值.
* type -将用户输入的值转化为哪种类型.
* choices - 参数可输入值的范围或选择.
* required - 命令行输入的值是否可以被忽略（布尔量）.
* help - 参数的简要描述.
* metavar - useage中显示的参数的名称.
* dest - 要添加到解析参数返回的对象中的属性的名称.
'''

def register_runjob_command():
    """注册runjob命令到Flask CLI"""
    @app.cli.command("runjob")
    @click.option("-m", "--name", required=True, help="指定job名")
    @click.option("-a", "--act", help="Job动作")
    @click.option("-p", "--param", multiple=True, help="业务参数")
    def runjob(name, act, param):
        """运行定时任务"""
        ret_params = {
            'name': name,
            'act': act,
            'param': list(param) if param else []
        }

        if not ret_params['name']:
            tip_msg = '''
                请正确调度Job
                flask runjob -m Test  (  jobs/tasks/Test.py )
                flask runjob -m test/Index (  jobs/tasks/test/Index.py )
            '''
            app.logger.info(tip_msg)
            click.echo(tip_msg)
            return False

        module_name = ret_params['name'].replace("/", ".")
        try:
            import_string = "from jobs.tasks.%s import JobTask as  job_target" % (module_name)
            exec(import_string, globals())
            target = job_target()
            target.run(ret_params)
        except Exception as e:
            traceback.print_exc()
            click.echo(f"执行任务失败: {str(e)}")
            return False