python - 如何将 APScheduler JobStore 与 SQLAlchemy 模型(外键)相关联 - Python Flask

Question

我有一个使用 flask.ext.sqlalchemy 和 apscheduler.schedulers.background 的 Python Flask 应用程序。我创建了一个 JobStore 并获得了一个名为 apscheduler_jobs 的表，它包含以下字段:

|id  |next_run_time|job_state|
------------------------------
|TEXT|   REAL      |  TEXT   |

我想使用如下方式将 SQLAlchemy 模型对象关联到该表:

from apscheduler.schedulers.background import BackgroundScheduler
scheduler = BackgroundScheduler()
scheduler.add_jobstore('sqlalchemy', url=app.config['SQLALCHEMY_DATABASE_URI'])

class Event(db.Model):

    __tablename__ = "event"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    name = db.Column(db.String(255), nullable=False)
    jobs = db.relationship('scheduler', backref='apscheduler_jobs')

所以我想使用 APScheduler apscheduler_jobs 中的表，然后将其与我的 Event 对象的外键相关联。最后一行基本上会中断，因为“调度程序”不是定义的 SQLAlchmey 模型

qlalchemy.exc.InvalidRequestError: When initializing mapper Mapper|Event|event, expression 'scheduler' failed to locate a name ("name 'scheduler' is not defined"). If this is a class name, consider adding this relationship() to the <class 'project.models.Event'> class after both dependent classes have been defined.

所以我认为我需要一个名为“job”或其他东西的中间模型类，然后将其与 apscheduler_jobs 相关联，但这里的某些东西仍然感觉很糟糕 - 因为 APScheduler 正在制作这个表我无法控制那里发生的事情- 我应该担心吗？

编辑 1:所以我创建了 2 个模型，一个“事件”然后一个“工作”，然后“工作”与表 apscheduler_jobs

相关

class Job(db.Model):

    __tablename__ = "job"
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    name = db.Column(db.String(255), nullable=False)
    apscheduler_job_id = db.Column(db.Integer, db.ForeignKey('apscheduler_jobs.id'))
    event_id = db.Column(db.Integer, db.ForeignKey('event.id'))

问题是，当我删除数据库并重新创建它时，它抛出了错误:

sqlalchemy.exc.NoReferencedTableError:与“job.apscheduler_job_id”列关联的外键找不到用于生成目标列“id”的外键的表“apscheduler_jobs”

现在我可以在我的数据库创建脚本中解决这个问题，但我还是觉得我做错了

编辑2我设法让它工作，虽然这感觉很不对，但我现在有 3 个模型:Event、Job 和 APSchedulerJobsTable。最终模型基本上与 APScheduler apscheduler_jobs 的外观相匹配。不过，必须有更好的方法来做到这一点。

from project import db


class Event(db.Model):

    __tablename__ = "event"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    name = db.Column(db.String(255), nullable=False)
    jobs = db.relationship('Job', backref='job_event')


class Job(db.Model):

    __tablename__ = "job"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    apscheduler_job_id = db.Column(db.TEXT, db.ForeignKey('apscheduler_jobs.id'))
    event_id = db.Column(db.Integer, db.ForeignKey('event.id'))


class APSchedulerJobsTable(db.Model):
    # TODO: This feels bad man
    __tablename__ = "apscheduler_jobs"

    id = db.Column(db.TEXT, primary_key=True, autoincrement=True)
    next_run_time = db.Column(db.REAL)
    job_state = db.Column(db.TEXT)

Answer 1

好吧，两个解决方案——在我看来都不是很完美:

解决方案一，可能更简洁 - 只需在包含 aspscheduler_job_ids 的作业表中有一个文本字段 - 虽然这不是外键，但一旦知道 aspscheduler_job ID，就可以继续并保存在job表中，以备后用

class Event(db.Model):

    __tablename__ = "event"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    name = db.Column(db.String(255), nullable=False)
    jobs = db.relationship('Job', backref='job_event')


class Job(db.Model):

    __tablename__ = "job"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    event_id = db.Column(db.Integer, db.ForeignKey('event.id'))
    apscheduler_job_id = db.Column(db.TEXT)

捕获这个是为了删除完整的数据库，您需要运行它以包括删除非托管表 apscheduler_jobs:

db.reflect()
db.drop_all()

方案二，将apscheduler表添加到model本身，然后设置外键:

class Event(db.Model):

    __tablename__ = "event"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    name = db.Column(db.String(255), nullable=False)
    jobs = db.relationship('Job', backref='job_event')


class Job(db.Model):

    __tablename__ = "job"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    event_id = db.Column(db.Integer, db.ForeignKey('event.id'))
    apscheduler_job_id = db.Column(db.TEXT, db.ForeignKey('apscheduler_jobs.id'))


class APSchedulerJobsTable(db.Model):
    # TODO: This feels bad man
    __tablename__ = "apscheduler_jobs"

    id = db.Column(db.TEXT, primary_key=True, autoincrement=True)
    next_run_time = db.Column(db.REAL)
    job_state = db.Column(db.TEXT)
    job = db.relationship('Job', backref='job_event')