AutoLine源码分析之调度管理器

什么是AutoLine开源平台

AutoLine开源平台是一个开源自动化测试解决方案，基于RobotFramework进行二次开发，支持RobotFramework几乎所有的库。

def setup(self):
    self.scheduler = BackgroundScheduler({
        # 设置调度任务存储mysql连接串
        'apscheduler.jobstores.default': {
            'type': 'sqlalchemy',
            'url': self.app.config["TRIGGER_DATABASE_URL"]
        },
        # 设置调度执行器进程池信息
        'apscheduler.executors.processpool': {
            'type': 'threadpool',
            'max_workers': '30'
        },
        # 设置调度其他配置
        'apscheduler.job_defaults.coalesce': 'false',
        'apscheduler.job_defaults.max_instances': '4',
        'apscheduler.timezone': 'UTC',
    })

def load_job_list(self):
    with self.app.app_context():
        # 查询所有项目
        projects = AutoProject.query.all()
        
        for p in projects:
            if p.enable and self.scheduler.get_job(p.id) is None:
                cron = p.cron.replace("\n", "").strip().split(" ")
                # 判断cron表达式是否有效
                if len(cron) < 5:
                    continue
                # 新增任务
                j = self.scheduler.add_job(func=run_job, 
                    trigger='cron', 
                    name=p.name, 
                    replace_existing=True,
                    minute=cron[0], 
                    hour=cron[1], 
                    day=cron[2], 
                    month=cron[3],
                    day_of_week=cron[4],
                    id="%s" % p.id, 
                    args=(p.id,))
            else:
                self.update_job(p.id)