flask_apscheduler源码分析

前言

遵循flask框架的标准的库，都称为flask扩展，flask_apscheduler是对apscheduler的扩展，也称为flask的扩展，最近使用flask_apscheduler遇到了一个job死亡的bug。现象：job平时是正常启动的，突然某个时刻全部挂了。。

环境介绍

官方文档：https://viniciuschiele.github.io/flask-apscheduler/

当前分析版本：1.12.4

安装方式：pip install Flask-APScheduler

源码位置：site-packages目录下

包结构介绍

flask_apscheduler是个包模块，包括__init__.py，共计6个模块

加载顺序

from flask_apscheduler import APScheduler

一般情况下，我们会在flask程序中，写下如上一行，此时flask_apscheduler的__init__.py中没有缩进的代码开始执行

init.py分析

from apscheduler.schedulers.base import STATE_PAUSED, STATE_RUNNING, STATE_STOPPED
from .scheduler import APScheduler

主要做了两件事

1、从标准库apscheduler下的base模块中，导入几个全局变量

2、从当前包下的scheduler模块中导入APScheduler类

scheduler模块分析

我们主要分析的是flask_apscheduler包模块下的scheduler.py模块

看了下这个scheduler.py模块共计400多行，我觉得还行

scheduler分析过程一：模块导入

import flask
import functools
import logging
import socket
import warnings
import werkzeugfrom apscheduler.events import EVENT_ALL
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.base import JobLookupError
from flask import make_response
from . import api
from .utils import fix_job_def, pop_trigger

总体的导入分3部分

1、标准库的导入

functools、logging、socket、warnings、apscheduler（重点依赖这个标准库）

2、第三方库

flask、werkzeug

3、自己写的模块

api、utils

整体说明：作者同时使用了标准库、比如logging用于日志打印的标准库，还有地方依赖库，当然是flask和werkzeug（flask依赖的底层网络库）、还有自己写的两个模块，api和utils。。

最最最重要的apscheduler的使用，尤其是导入BackgroundScheduler这个类

scheduler分析过程二：创建日志分析对象

LOGGER = logging.getLogger('flask_apscheduler')

scheduler分析过程三：创建APScheduler类

class APScheduler(object):
…………省略…………

这个APScheduler创建的对象，是以后我们经常用的对象，作为整个模块的业务逻辑入口，后续单独开篇文章介绍这个类的封装。

初步总结

scheduler就干了3件事、导入模块、创建日志分析对象、创建APScheduler类。

继续分析当前包模块

上面已经分析了__init__.py模块、还有scheduler.py模块，还记得scheduler.py下面这两句代码吗？

from . import api

from .utils import fix_job_def, pop_trigger

我们将继续分析api模块和utils模块，因为这俩模块先后加载到内存中了

api模块分析

scheduler.py模块加载的时候，导入了api.py模块，此时api.py模块没有缩进代码将会被执行

api模块分析过程一：模块导入

import loggingfrom apscheduler.jobstores.base import ConflictingIdError, JobLookupError
from collections import OrderedDict
from flask import current_app, request, Response
from .json import jsonify

过程也是3部分

1、导入标准库（导入过的不会重复导入，所以这里写了也没事，内存中是同一个模块对象）

logging、apscheduler、collections模块

2、导入第三方库

flask

3、导入自己写的模块

json

api模块分析过程二：创建函数

1、连续创建了9个函数对象

2、且他们都与flask应用对象有所关联，我给找其中一个函数给大伙看看

def add_job():"""Adds a new job."""data = request.get_json(force=True)try:job = current_app.apscheduler.add_job(**data)return jsonify(job)except ConflictingIdError:logging.warning(f'Job {data.get("id")} already exists.')return jsonify(dict(error_message='Job %s already exists.' % data.get('id')), status=409)except Exception as e:logging.error(e, exc_info=True)return jsonify(dict(error_message=str(e)), status=500)

add_job，通过找个函数我们随时向调度器中添加一个job，可以说是一种动态添加job的方式！！

current_app 表示当前flask对象

current.apscheduler表示与之关联的Scheduler对象

return jsonify(job) 最终竟然也返回了一个响应，这是为啥呢？原来是flask_apscheduler给我们留的后门！！

在Scheduler类中，有个方法，是在框架中唯一使用这些api模块中的函数的地方

开关在这里，原来我们可以通过SCHEDULER_API_ENABLED，这样的flask配置修改是否开启快捷开关，这里不看源码，是肯定不知道有这个后门的，看来我也要开启了

初步总结

api模块中的函数，可以在当前flask应用注册路由，那样我们通过http请求，就能操作job了，非常的方便debug呀，爽..

utils模块分析

这个模块，看名字就知道是工具模块了，我们看看这个模块加载的时候干了什么

utils.py模块分析过程一：模块导入

import dateutil.parser
import sixfrom apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.date import DateTrigger
from apscheduler.triggers.interval import IntervalTrigger
from collections import OrderedDict

1、标准库

collections

apscheduler

2、第三方库

dateutil

six

utils.py模块分析过程二：创建几个函数

作者真是代码写的干净利索啊，牛逼，这几个函数要工具相关，比如job转为字典，看来是来兜底用的模块，厉害，抽空看看几个函数具体是干啥的

json模块分析

json模块分析过程一：模块导入

from __future__ import absolute_importimport datetime
import flaskfrom apscheduler.job import Job
from .utils import job_to_dictimport json  # noqa

1、标准库

__future__

datetime

apscheduler

json

2、三方库

flask

看来这个模块主要是操作json格式的

json模块分析过程二：创建全局变量

loads = json.loads

拿来注意体现的好啊，创建一个loads全局变量，指向的是json模块下的loads函数，这样以后用这个函数就轻松了……

json模块分析过程三：创建函数

创建的dumps函数和jsonify函数

json模块分析过程四：创建类

class JSONEncoder(json.JSONEncoder):def default(self, obj):if isinstance(obj, datetime.datetime):return obj.isoformat()if isinstance(obj, Job):return job_to_dict(obj)return super(JSONEncoder, self).default(obj)

创建了一个用于编解码json的类