特定时间之后，自动更新烧瓶-Sqlalchemy

Question

我有一个类似的DB模型：

class Payment(db.Model):
    id = db.Column(db.Integer(), primary_key=True)
    user_id = db.Column(db.Integer(), db.ForeignKey('user.id'))
    ticket_status = db.Column(db.Enum(TicketStatus, name='ticket_status', default=TicketStatus.UNUSED))
    departure_time = db.Column(db.Date)

我想在datetime.utcnow()通过departure_time的日期值之后从所有ticket_status更改值。

我尝试这样的编码：

class TicketStatus(enum.Enum):
    UNUSED = 'UNUSED'
    USED = 'USED'
    EXPIRED = 'EXPIRED'
    def __repr__(self):
        return str(self.value)

class Payment(db.Model):
    id = db.Column(db.Integer(), primary_key=True)
    user_id = db.Column(db.Integer(), db.ForeignKey('user.id'))
    ticket_status = db.Column(db.Enum(TicketStatus, name='ticket_status', default=TicketStatus.UNUSED))
    departure_time = db.Column(db.Date)
    # TODO | set ticket expirations time
    def __init__(self):
        if datetime.utcnow() > self.departure_time:
            self.ticket_status = TicketStatus.EXPIRED.value
        try:
            db.session.add(self)
            db.session.commit()
        except Exception as e:
            db.session.rollback()

我也尝试过这样的尝试：

def ticket_expiration(self, payment_id):
    now = datetime.utcnow().strftime('%Y-%m-%d')
    payment = Payment.query.filter_by(id=payment_id).first()
    if payment.ticket_status.value == TicketStatus.USED.value:
        pass
    elif payment and str(payment.departure_time) < now:
        payment.ticket_status = TicketStatus.EXPIRED.value
    elif payment and str(payment.departure_time) >= now:
        payment.ticket_status = TicketStatus.UNUSED.value
    try:
        db.session.commit()
    except Exception as e:
        db.session.rollback()
    return str('ok')

，但是当datetime.utcnow()通过departure_time的日期值时，似乎没有影响。

所以我的问题的目的是，如何在一组次之后自动从行更改值。？

Answer 1

最后，我使用 flask_apscheduler 来弄清楚这一点，这是我代码的片段解决了此问题：

安装 flask_apscheduler ：

pip3 install flask_apscheduler

创建新模块 tasks.py

from datetime import datetime
from flask_apscheduler import APScheduler
from app import db
from app.models import Payment, TicketStatus
scheduler = APScheduler()

def ticket_expiration():
    utc_now = datetime.utcnow().strftime('%Y-%m-%d')
    app = scheduler.app
    with app.app_context():
        payment = Payment.query.all()
        for data in payment:
            try:
                if data.ticket_status.value == TicketStatus.USED.value:
                    pass
                elif str(data.departure_time) < utc_now:
                    data.ticket_status = TicketStatus.EXPIRED.value
                elif str(data.departure_time) >= utc_now:
                    data.ticket_status = TicketStatus.UNUSED.value
            except Exception as e:
                print(str(e))
            try:
                db.session.commit()
            except Exception as e:
                db.session.rollback()
    return str('ok')

，然后在 __ INIT __。

def create_app(config_class=Config):
    app = Flask(__name__)
    app.config.from_object(Config)
    # The other packages...
    # The other packages...
    scheduler.init_app(app)
    scheduler.start()
    return app
# import from other_module...
# To avoid SQLAlchemy circular import, do the import at the bottom.
from app.tasks import scheduler 

这是针对 config.py ：

class Config(object):
    # The others config...
    # The others config...
    # Flask-apscheduler
    JOBS = [
        {
            'id': 'ticket_expiration',
            'func': 'app.tasks:ticket_expiration',
            'trigger': 'interval',
            'hours': 1, # call the task function every 1 hours
            'replace_existing': True
        }
    ]
    SCHEDULER_JOBSTORES = {
        'default': SQLAlchemyJobStore(url='sqlite:///flask_context.db')
    }
    SCHEDULER_API_ENABLED = True

在上面的配置中，我们可以按照我们的情况调用每1小时，秒或其他时间更新DB的功能，以获取更多信息，以设置我们可以在此处看到的间隔时间。

我希望这个答案可以帮助将来面对这一点的人。

Answer 2

您可以用"二手"列替换状态列，该列将包含布尔值并为状态制造混合属性。https://docs.sqlalchemy.org/en/13/orm/extensions/hybrid.html

class Payment(db.Model):
    id = db.Column(db.Integer(), primary_key=True)
    user_id = db.Column(db.Integer(), db.ForeignKey('user.id'))
    used = db.Column(db.Boolean(), default=False)
    departure_time = db.Column(db.Date)
    @hybrid_property
    def status(self):
        if datetime.utcnow() > self.departure_time:
            return "EXPIRED"
        elif self.used:
            return "USED"
        return "UNUSED"