FastAPI:在特定时间执行动作

问题描述 投票:0回答:0

我在系统上工作,提醒用户一些动作,比如警报,所以我需要让它像触发一些功能一样工作,从数据库(postgresql)中选择数据并通过 FCM 通知用户。最简单的解决方案是制作可重复的任务,这将在短时间内完成,但我需要及时通知。不是之前也不是之后。所以,我的问题如下:“如果提醒日期时间存储在数据库中并且提醒可以重复(重复和间隔的存储类型。没有下一个日期时间),我该怎么做?”。

提醒模式如下:

class Reminder(Base):

    __tablename__ = 'reminders'


    ...
    date = Column(DateTime, nullable=False)
    last_reminded_at = Column(DateTime, nullable=True)
    repeat_end_date = Column(DateTime, nullable=True)  # None - infinite.
    category = Column(Enum(ReminderCategoryEnum), nullable=False)
    repeat_type = Column(
        Enum(ReminderRepeatTypeEnum),
        default=ReminderRepeatTypeEnum.NEVER,
        server_default=ReminderRepeatTypeEnum.NEVER.name,
        nullable=False,
    )
    repeat_interval = Column(RelativeInterval(native=True), nullable=False)  # Interval
    is_finished = Column(Boolean, default=False, server_default=false(), nullable=False)
    is_notified = Column(Boolean, default=False, server_default=false(), nullable=False)

    def __init__(self, *args, **kwargs):  # noqa: D107
        repeat_type: Optional[ReminderRepeatTypeEnum] = kwargs.get('repeat_type')
        if 'repeat_interval' not in kwargs:
            if not repeat_type:
                repeat_type = ReminderRepeatTypeEnum.NEVER
            self.repeat_interval = repeat_type.timedelta
        super().__init__(*args, **kwargs)

    @property
    def previous_date(self) -> Optional[datetime.datetime]:
        ...

    @property
    def next_date(self) -> Optional[datetime.datetime]:
        ...
python sqlalchemy scheduled-tasks fastapi
© www.soinside.com 2019 - 2024. All rights reserved.