0

当我使用flask-apscheduler（不是apscheduler）时，我的flask web 项目出现了一些问题。特别是当我使用db(flask-sqlalchemy)对象时。问题可能是：

JOBS = [
        {
            'id': 'job1',
            'func': 'app.monitor.views:test',
            'args': (),
            'trigger': 'interval',
            'seconds': 2
        }
    ]

./app/初始化.py：

from flask import Flask
from flask.ext.bootstrap import Bootstrap
from flask.ext.mail import Mail
from flask.ext.moment import Moment
from flask.ext.sqlalchemy import SQLAlchemy
from flask.ext.login import LoginManager
from flask.ext.pagedown import PageDown
from flask_apscheduler import APScheduler
from celery import Celery

# from apscheduler.schedulers.blocking import BlockingScheduler
from config import config,Config
bootstrap = Bootstrap()
mail = Mail()
moment = Moment()
db = SQLAlchemy()
pagedown = PageDown()
celery = Celery(__name__, broker=Config.CELERY_BROKER_URL)
# https://pypi.python.org/pypi/Flask-APScheduler
scheduler = APScheduler()
login_manager = LoginManager()
login_manager.session_protection = 'strong'
login_manager.login_view = 'auth.login'

def create_app(config_name):
    app = Flask(__name__)
    app.config.from_object(config[config_name])
    config[config_name].init_app(app)
    bootstrap.init_app(app) 
    mail.init_app(app)
    moment.init_app(app)
    db.init_app(app)
    login_manager.init_app(app)
    pagedown.init_app(app)
    scheduler.init_app(app)
    celery.conf.update(app.config)
    if not app.debug and not app.testing and not app.config['SSL_DISABLE']:
        from flask.ext.sslify import SSLify
        sslify = SSLify(app)
    from .monitor import monitor as monitor_1_0_blueprint
    from .laser import laser as laser_1_0_blueprint
    app.register_blueprint(monitor_blueprint,url_prefix='/monitor/api')
    app.register_blueprint(laser_1_0_blueprint,url_prefix='/laser/api/v1.0')
    return app

错误1：db是：错误2：db是：找不到记录器“apscheduler.executors.default”的处理程序错误3：db是：引发RuntimeError（'在应用程序上下文之外工作'）RuntimeError：在应用程序上下文之外工作

问题的关键是在flask-apscheduler作业函数（views.py）中获取db和app对象：

from app import scheduler
def test():
    #to Solve the log error problem
    import logging
    log = logging.getLogger('apscheduler.executors.default')
    log.setLevel(logging.INFO)  # DEBUG
    fmt = logging.Formatter('%(levelname)s:%(name)s:%(message)s')
    h = logging.StreamHandler()
    h.setFormatter(fmt)
    log.addHandler(h)
    #get the app object
    app = scheduler.app
    #get the db object and use it
    with app.app_context():
        print '........................',db
4

2 回答 2

0 
def test():
    #to Solve the log error problem
    import logging
    log = logging.getLogger('apscheduler.executors.default')
    log.setLevel(logging.INFO)  # DEBUG
    fmt = logging.Formatter('%(levelname)s:%(name)s:%(message)s')
    h = logging.StreamHandler()
    h.setFormatter(fmt)
    log.addHandler(h)
    #get the app object
    app = scheduler.app
    #get the db object and use it
    with app.app_context():
        print '........................',db        #the right db object
于 2016-07-23T21:45:42.083 回答
0 
    from app import scheduler#
    def test():
        #to Solve the log error problem
        import logging
        log = logging.getLogger('apscheduler.executors.default')
        log.setLevel(logging.INFO)  # DEBUG
        fmt = logging.Formatter('%(levelname)s:%(name)s:%(message)s')
        h = logging.StreamHandler()
        h.setFormatter(fmt)
        log.addHandler(h)
        #get the app object
        app = scheduler.app
        #get the db object and use it
        with app.app_context():
        print '........................',db
于 2016-07-23T21:53:10.527 回答