How to create periodic tasks with Celery 4 and Celery 5

Let's analyze the usage of periodic tasks as a useful example. Our task is to get information about a free ebook from Packt Publishing (it could be any other site). We'd like to have an email every day at 7.00 AM with the title of this book.

Let's look at how to do it with Celery (both 4 and 5 versions) and Django. In both cases 'sd' is the name of my Django's project. I've separated my settings into a few parts and 'settings.production' part is information that I've got a 'production.py' file in the 'settings' directory.

For Celery 4:

1. In the directory of the Django's project, we should create a standard celery.py file:

                import os
                from celery import Celery

                os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'settings.production')

                app = Celery('sd')
                app.config_from_object('django.conf:settings', namespace='CELERY')
                app.autodiscover_tasks()
              

2. In the __init__.py of the project we should add this one line:

                  __all__ = ['celery_app']
              

3. Inside one of the applications we should create tasks.py file:

                    import requests
                    from bs4 import BeautifulSoup
                    from django.core.mail import EmailMessage
                    from celery.schedules import crontab
                    from celery.task import periodic_task

                    @periodic_task(run_every=(crontab(minute=0, hour=7)),
                                   name='task_send_email_about_ebook',
                                   ignore_result=True
                                   )
                    def task_send_email_about_ebook():
                        page = requests.get('https://www.packtpub.com/free-learning')
                        soup = BeautifulSoup(page.content, 'html.parser')
                        book_title = soup.find('h3', attrs={'class': 'product-info__title'}).text.split('-', 1)[1].strip()
                        subject = "Your free e-book from PacktPub is available!"
                        message = f"Your new e-book is '{book_title}'. \n"
                        message += f"Book is available at <a href='https://www.packtpub.com/packt/offers/free-learning'>Free Learning</a>."
                        email = EmailMessage(subject,
                                             message,
                                             'from@domaine.com',
                                             ['to@somedomain.com'])
                        email.send(fail_silently=False)
              

For Celery 5:

1. In the directory of the Django's project, we should create a standard celery.py file:

                    import os
                    from celery import Celery

                    # set the default Django settings module for the 'celery' program.
                    os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'settings.production')

                    app = Celery('sd')

                    app.config_from_object('django.conf:settings', namespace='CELERY')
                    app.autodiscover_tasks()
              

2. In the __init__.py of the project we should add this one line:

from .celery import app as celery_app

3. Inside one of the applications we should create tasks.py file:

                    import requests
                    from bs4 import BeautifulSoup
                    from celery.schedules import crontab
                    from django.core.mail import EmailMessage

                    from sd.celery import app

                    @app.task
                    def task_send_email_about_ebook():
                        page = requests.get('https://www.packtpub.com/free-learning')
                        soup = BeautifulSoup(page.content, 'html.parser')
                        book_title = soup.find('h3', attrs={'class': 'product-info__title'}).text.split('-', 1)[1].strip()
                        subject = "Your free e-book from PacktPub is available!"
                        message = f"Your new e-book is '{book_title}'. \n"
                        message += f"Book is available at <a href='https://www.packtpub.com/packt/offers/free-learning'>Free Learning</a>."
                        email = EmailMessage(subject,
                                             message,
                                             'from@domaine.com',
                                             ['to@somedomain.com'])
                        email.send(fail_silently=False)

                    app.conf.beat_schedule = {
                        "task_send_email_about_ebook": {
                            "task": "frontend.tasks.task_send_email_about_ebook",
                            "schedule": crontab(hour=7, minute=0)
                            }
                        }
              

Conclusion

Setting up periodic tasks is essential for many web applications, from sending newsletters to daily database cleanups. Celery makes this process robust and scalable.