Search code examples
pythonscheduled-taskspyramid

Python Pyramid periodic task


My question is about how to perform a simple task in Pyramid Python framework?

I have a table with users birthdays and I'd like to check daily if someone's birthday is coming, and if so, print a simple greeting.

So I've defined some stuff in my views.py:

def print_greetings():
   tomorrow = datetime.date.today()+datetime.timedelta(days=1)
    users = sess.query(User).filter(func.DATE_FORMAT(User.bday, "%d_%m")==tomorrow.strftime("%d_%m")).all()
    if len(users) > 0:
        for u in users:
            print("Happy Birthday", u.name)

So the question is: How and where should I write something for execute this function once a day?

UPD: I need to run the task from the Pyramid app, external tools like cron are not the thing i'm looking for.

Maybe some sort of timer which will execute print_greetings()? And after execution will start a new instance of timer?


Solution

  • While Celery is a rather "heavy-weight" solution for a task like this, I'd recommend using APScheduler. You can setup the scheduler exactly like in cron, but it will run within your app. In this case the in-memory job store works fine.

    import atexit
    from apscheduler.schedulers.background import BackgroundScheduler
    
    scheduler = BackgroundScheduler()
    scheduler.start()
    scheduler.add_job(print_greetings,
        id='greetings', 
        name='Send out birthday greetings', 
        trigger='cron', 
        minute=0, 
        hour=12,
    )
    
    atexit.register(lambda: scheduler.shutdown())