Cron Jobs With Python

Manage Cron Jobs With Python

A simple scenario. You need a program to run every time when you reboot. Well, Cron is a solution in one of many. But to manage Cron jobs we often had to deal with the crontab
A simple scenario. You need a program to run every time when you reboot. Well, Cron is a solution in one of many. But to manage Cron jobs we often had to deal with the crontab

Cron is a software utility which is a time-based scheduler in Unix-like OS. We use Cron to schedule jobs to run periodically at fixed times, dates, or intervals. Which means you can run a process every month or day or even year. Even every minute if you want and more. Which Cron will take care of and we don’t even have to think about it.

A simple scenario. You need a program to run every time when you reboot. Well, Cron is a solution in one of many. But to manage Cron jobs we often had to deal with the crontab. Which seems a little bit hard. Although we have many possible solutions. And we are going to talk about one of them today. Yes, that’s Python. It’s a great language that comes in handy in situations like that. We will learn today how to create a Cronjob with a different time schedule, how to monitor them, and how to remove them all or one by one. Alright, let’s get started.

Prerequisites

  • Linux (mine is ubuntu 18.04)
  • Python3
  • pip
  • venv

Environment Setup & Installation

Always remember, whenever you are becoming a crazy horse to build a new python project, just hold that horse for a few seconds. Set up a virtual environment first and thank me later. Go to your favorite directory and follow the steps

python3 -m venv my_env
cd my_env

Now we need to activate the virtual environment. And then we need to install python-crontab.

. bin/activate
pip install python-crontab

Now create our project repository. We will call our project schedule_manager.

mkdir schedule_manager

Let’s Play Before Education

Quick Start

Before we start, let’s see some demo. You can do that by creating any python file inside the project. First import Crontab module.

from crontab import CronTab

First, the CronTab class is used to instantiate a cron object

cron = CronTab(user=True)

And to create a new job,

job = cron.new(command='python example.py')

Now if you want to execute the job every minute,

job.minute.every(1)

Lastly, write the cron.

cron.write()

 

More Periodic Cycle

And that’s all. What if we need different types of schedules. It could be hourly, monthly, or any periodic cycle. Check the below code.

# For hourly cronjob,

job.minute.on(0)
job.hour.during(0, 23)

# For daily cronjob,

job.minute.on(0)
job.hour.on(0)

# For weekly cronjob,

job.minute.on(0)
job.hour.on(0)
job.dow.on(1)

# For monthly cronjob,

job.minute.on(0)
job.hour.on(0)
job.day.on(1)
job.month.during(1, 12)

# For yearly cronjob,

job.minute.on(0)
job.hour.on(0)
job.month.on(12)

# Even if you want to run a cronjob after every reboot,

job.every_reboot()

 

Let’s Develop Our Schedule Manager

Application Structure

.
├── app
│   ├── __init__.py
│   ├── run.py
│   ├── scheduler.py
│   └── worker.py
├── requirements.txt
└── test.txt

In our app/scheduler.py file, we will have the Scheduler class where we will write all the necessary methods to interact with cronjobs.

The app/run.py file will be used to execute them.

And lastly, app/worker.py is the file that our cronjob will execute. Inside the file, you can write any instructions to the computer. For example, writing a file, sending a desktop notification, automated skype message, or anything. We will write a file which is the test.txt file in our case to understand if the cronjob worked or not.

 

Develop Scheduler Class

In our app/scheduler.py file, Let’s import the Crontab module and start writing the Scheduler class.

"""
    ----------------------------------------------------------------------------
    Scheduler module
    ----------------------------------------------------------------------------
    1. Can create cron jobs with specific time schedule

        Case           Meaning
        @reboot            Every boot
        @hourly            0 * * * *
        @daily         0 0 * * *
        @weekly            0 0 * * 0
        @monthly       0 0 1 * *
        @yearly            0 0 1 1 *
    ----------------------------------------------------------------------------
    2. can remove all cron jobs
    ----------------------------------------------------------------------------
"""
from crontab import CronTab, CronItem

class Scheduler:
    """Scheduler class"""
    def __init__(self, command: str):
        """
        Here we will initiate a Cron
        and set the command
        :param command: command to execute
        """
        self.cron = CronTab(user=True)
        self.command = command

 

Our __init__ method takes command as a parameter. Which we will set in our job. We also instantiate a cron object in our __init__method.

Now let’s write a method to actually create a job.

def __create_cron_job(self) -> CronItem:
    """
    Create a new job and return it
    :return: job
    """
    job = self.cron.new(command=self.command)
    return job

After that, we will add methods to execute jobs in many periodic cycles like, minutely, hourly, daily, weekly, monthly, yearly, and after every reboot.

def minutely_execution(self) -> bool:
    """
    Execute job every minute
    :return:
    """
    try:
        job = self.__create_cron_job()
        job.minute.every(1)
        job.enable()
        self.cron.write()
        if self.cron.render():
            return True
        return False
    except Exception as error:
        print(error)
        return False

def hourly_execution(self) -> bool:
    """
    Execute job every hour
    :return:
    """
    try:
        job = self.__create_cron_job()
        job.minute.on(0)
        job.hour.during(0, 23)
        job.enable()
        self.cron.write()
        if self.cron.render():
            return True
        return False
    except Exception as error:
        print(error)
        return False

def daily_execution(self) -> bool:
    """
    Execute job every day
    :return:
    """
    try:
        job = self.__create_cron_job()
        job.minute.on(0)
        job.hour.on(0)
        job.enable()
        self.cron.write()
        if self.cron.render():
            return True
        return False
    except Exception as error:
        print(error)
        return False

def weekly_execution(self) -> bool:
    """
    Execute job every week
    :return:
    """
    try:
        job = self.__create_cron_job()
        job.minute.on(0)
        job.hour.on(0)
        job.dow.on(1)
        job.enable()
        self.cron.write()
        if self.cron.render():
            return True
        return False
    except Exception as error:
        print(error)
        return False

def monthly_execution(self) -> bool:
    """
    Execute job every month
    :return:
    """
    try:
        job = self.__create_cron_job()
        job.minute.on(0)
        job.hour.on(0)
        job.day.on(1)
        job.month.during(1, 12)
        job.enable()
        self.cron.write()
        if self.cron.render():
            return True
        return False
    except Exception as error:
        print(error)
        return False

def yearly_execution(self) -> bool:
    """
    Execute job every year
    :return:
    """
    try:
        job = self.__create_cron_job()
        job.minute.on(0)
        job.hour.on(0)
        job.month.on(12)
        job.enable()
        self.cron.write()
        if self.cron.render():
            return True
        return False
    except Exception as error:
        print(error)
        return False

def execute_after_reboot(self) -> bool:
    """
    Execute job after every reboot
    :return:
    """
    try:
        job = self.__create_cron_job()
        job.every_reboot()
        job.enable()
        self.cron.write()
        if self.cron.render():
            return True
        return False
    except Exception as error:
        print(error)
        return False

We will also write a method to destroy all cronjobs in case we mess things up.

def delete_all_jobs(self) -> bool:
    """
    remove all cron jobs
    :return:
    """
    try:
        self.cron.remove_all()
        return True
    except Exception as error:
        print(error)
        return False

The Worker In Our Cron Job

Well just test our cronjob we will add some file writing code in our app/worker.py file. Which is responsible for writing in the test.txt file.

from datetime import datetime


def worker(file_path: str):
    """
    This worker could be anything. Any work you desire to be executed
    :param file_path: path of the file we wanna write
    """
    with open(file_path, "a") as file:
        file.write('\nWorker1 executed at ' + str(datetime.now()))
        file.close()


if __name__ == "__main__":
    path = '/home/your-location/' \
           'schedule_manager/test.txt'

    worker(path)

 

Running our program

In our app/run.py file, write the below code to create a new cronjob.

from app.scheduler import Scheduler


if __name__ == "__main__":
    python_location = '/home/your-location/bin/python3'
    file = '/worker.py'
    cron_job_command = python_location + " " + file

    # Execute job
    scheduler = Scheduler(cron_job_command)
    scheduler.minutely_execution()

Which will run the cron_job_command in every minute. Which is basically the execution of a python file.

 

What else can one do?

Let’s hold back your interest for some time. What If I tell you, you can execute a Linux command in subprocess which will be responsible to restart your computer inside the app/worker.py instead of just writing some nonsense file? What if you set that command in cronjob after every reboot? Well, bad things will happen. Don’t try this at anyone’s home. I won’t be responsible. Explore more to know what else can you manage by this knowledge. Life will be a lot easier after that.

Stay safe and manage your repeated works using cronjob!

Github Repo Of this Article: https://github.com/zim0101/schedule_manager