sqlalchemy的Flask应用程序正在尝试使用FlaskAPScheduler计划作业

Question

它在执行flask应用程序时运行良好，但在flask ApsScheduler上下文中执行时出现以下错误： AttributeError: Neither 'InstrumentedAttribute' object nor 'Comparator' object associated with SMSMessage.logs has an attribute 'count'

我们对APScheduler类进行了子分类，因为我们觉得可以强制它使用flask上下文

我的模型定义为：

class SMSMessage(BaseModel):
__tablename__ = 'sms_message'

logs = relationship("SMSLog", backref='sms', lazy='dynamic')

@hybrid_property
def last_status(self):
    if self.logs.count() > 0:
        last_status = self.logs.order_by(SMSLog.date_created.desc()).first()
        return last_status.at_status_code
    return "NA"

作业功能在主功能文件中定义为：

import os
import dateutil.parser

from flask_apscheduler import APScheduler as _BaseAPScheduler

from app import settings
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from flask import Blueprint, Flask
from datetime import datetime

from app.database import db
from app.database.models import SMSMessage, SMSLog

class APScheduler(_BaseAPScheduler):
    def run_job(self, id, jobstore=None):
        with self.app.app_context():
            super().run_job(id=id, jobstore=jobstore)

application = Flask(__name__)

def run_queued_messages():
    with application.app_context():
        sms_messages = SMSMessage.query.filter(
            SMSMessage.last_status == 'Queued',
            SMSMessage.send_at <= datetime.now()
        ).all()

        for sms_message in sms_messages:
            print('do something')


def configure_app(flask_app):
    DB_HOST = os.getenv("DB_HOST")
    DB_PORT = os.getenv("DB_PORT")
    DB_DATABASE = os.getenv("DB_DATABASE")
    DB_USER = os.getenv("DB_USER")
    DB_PASSWORD = os.getenv("DB_PASSWORD")

    flask_app.config['SQLALCHEMY_DATABASE_URI'] = f'postgresql+psycopg2://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_DATABASE}'
    flask_app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = settings.SQLALCHEMY_TRACK_MODIFICATIONS
    flask_app.config['SCHEDULER_JOBSTORES'] = {
        'default': SQLAlchemyJobStore(
            url=flask_app.config['SQLALCHEMY_DATABASE_URI'],
            tableschema=os.getenv('SCHEMA_NAME')
        ),
    }
    # do not allow for api access job management
    flask_app.config['SCHEDULER_API_ENABLED'] = False
    flask_app.config['JOBS'] = [
        {
            'id': None,
            'func': run_queued_messages,
            'trigger': 'interval',
            'seconds': 1,
            'replace_existing': True
        }
    ]
    flask_app.config['SCHEDULER_EXECUTORS'] = {
        'default': {'type': 'threadpool', 'max_workers': 20}
    }
    flask_app.config['SCHEDULER_JOB_DEFAULTS'] = {
        'coalesce': False,
        'max_instances': 3
    }


def initialize_app(flask_app):
    configure_app(flask_app)

    db.init_app(flask_app)

initialize_app(application)
db.application = application

scheduler = APScheduler()
scheduler.init_app(application)
scheduler.start()

Answer 1

1. self.logs不是日志的集合，而是作为关系对Logs的引用。实际上，您需要首先获取计数
2. 当您实际上只需要知道是否有任何日志exists()时，使用count()是不好的做法。如果我们正在做饭，我问是否还有米饭，我不会指望你数一粒一粒的米饭，而只是检查锅里是不是空的
3. 如果不存在匹配项，.first()返回None。充分利用它：

@hybrid_property
def last_status(self):
    last_status = SMSLog.query\
        .filter(SMSLog.message_id == self.id)\
        .order_by(SMSLog.date_created.desc())\
        .first()
    return last_status.at_status_code if last_status is not None else "NA"