FastAPI: perform an action at a specific time

Question

I work on a system, that reminds the user about some action with an alarm.

I need to make it work by triggering some functions that select data from a database (PostgreSQL) and notify the user via FCM.

The easiest solution is to make a repeatable task, with a minimal interval, but I need to notify at the exact time, not before and not after it.

How can I do it, if the reminder datetime is stored in the database and reminders can be repeatable (store type of repeat and interval. Without next datetime)?

The Reminder model is the following:

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]:
        ...

Does this answer your question? [FastAPI python: How to run a thread in the background?](https://stackoverflow.com/questions/70872276/fastapi-python-how-to-run-a-thread-in-the-background) — Chris, Apr 25 '23 at 07:50
@Chris I guess, not. I'll use fastapi background tasks with wait until needed date and notify user, then wait until next date — Dmitriy Lunev, Apr 25 '23 at 13:13
@Chris yes. Unfortunately, my task is not async. My question was not about how to make my task work in background. I needed something like dynamic scheduler, that trigger some function, that notify user in some user-specified times (this time stores in database). I don't want to make a lot of queries. — Dmitriy Lunev, Apr 25 '23 at 13:22
I used `apscheduler` to manage background tasks and created task, that start at specific time and perform notification. Here is [my implementation of task recreate on server start up](https://stackoverflow.com/questions/76102199/recreate-background-tasks-on-server-restart?noredirect=1#comment134214547_76102199). — Dmitriy Lunev, Apr 27 '23 at 07:58
Using `apscheduler` was one of the solutions I was about to post here. Glad you got it resolved. I'll still try to post an answer though, once I get the chance, with all the available solutions, as this would help future readers. So please leave the question open — Chris, Apr 27 '23 at 09:12
@Chris Thank you for your answers. Sure, I' won't close question. — Dmitriy Lunev, Apr 28 '23 at 07:29

score 0 · Answer 1 · answered Jun 07 '23 at 13:01

We can implement a solution that leverages Redis Keyspace Notifications, which is a Pub/Sub for events affecting Redis data.

Solution Overview

When an Insert/Update event occurs on the Reminder table, a key is added to Redis with all the relevant info (ex. Reminder item id) and an expiration datetime = the relevant datetime that we want the reminder alarm to set off.
A Keyspace Notification Subscriber will be ready to consume any EXPIRE events and send the notification to the user.
In case of repeated notifications, the key should be added back to Redis with a new expiring date.
In case of a Reminder item deletion, Redis key can be removed (not critical, we can have the check inside the Subscriber code)

Pros:

Non-complicated solution based on a proven tool (Redis)
Easy to maintain and expand
In-time notification handling without mind-warping scheduling management.

Cons:

Complicates your stack as it adds Redis.
Need to think of connected cleanup between Reminder items and Redis items
Need to learn about Redis persistence and think about handling outages etc.

Let's see the parts:

On Reminder item Insert/Update to DB, we need to calculate the expiration time in seconds for the first notification
Insert the key to redis with the previously calculated expiration_time:
```
redis_conn.set(reminder_item.id, "reminder")
redis_con.expire(reminder_item.id, expiration_time_in_seconds)
```
redis_conn is a redis connection initiated as per the python instructions (initialization example in step 4 below)

Create an event handler method:

def event_habdler(msg):
    redis_reminder_key = msg["data"].decode("utf-8")

    # Use the redis_reminder_key to retrieve the corresponding   
    # reminder item from SQL and notify the user.

    # Check if you need to delete the key or update it's expiration time  
    if <check-for-last-notification>:
        redis_conn.delete(key)
    else:
        expiration_time_in_seconds = <calculate-next-notification-time-in-seconds>
        redis_conn.expire(redis_reminder_key, expiration_time_in_seconds)

Initialize a Redis connection on FastAPI startup and connect a Subscriber to the Keyspace Notification using FastAPI's startup event:

import redis

from fastapi import FastAPI

app = FastAPI()

@app.on_event("startup")
async def startup_event():
    redis_conn = redis.Redis(<connection-info>)
    redis_pubsub = redis_conn.pubsub()

    redis_conn.config_set('notify-keyspace-events', 'Ex')
    pubsub.psubscribe(**{"__keyevent@0__:expired": event_handler})
    pubsub.run_in_thread(sleep_time=0.01)

Suggestions for ease of use:

Create redis_conn in a separate file (like settings.py) and use it as settings.redis_conn
Optional: Add scheduled events for data sync checks between SQL and Redis.

Subscriber event_handler inspired from: https://medium.com/@imamramadhanns/working-with-redis-keyspace-notifications-x-python-c5c6847368a

FastAPI: perform an action at a specific time

1 Answers1