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.<\/p>\n
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.<\/p>\n
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<\/p>\n
python3 -m venv my_env<\/pre>\ncd my_env<\/pre>\nNow we need to activate the virtual environment. And then we need to install python-crontab.<\/strong><\/p>\n
. bin\/activate\r\npip install python-crontab<\/pre>\nNow create our project repository. We will call our project schedule_manager.<\/em><\/strong><\/p>\n
mkdir schedule_manager<\/pre>\nLet’s Play Before Education<\/h2>\n
Quick Start<\/h3>\n
Before we start, let’s see some demo. You can do that by creating any python file inside the project. First import Crontab module.<\/p>\n
from crontab import CronTab<\/pre>\nFirst, the CronTab class is used to instantiate a cron object<\/p>\n
cron = CronTab(user=True)<\/pre>\nAnd to create a new job,<\/p>\n
job = cron.new(command='python example.py')<\/pre>\nNow if you want to execute the job every minute,<\/p>\n
job.minute.every(1)<\/pre>\nLastly, write the cron.<\/p>\n
cron.write()<\/pre>\n<\/p>\n
More Periodic Cycle<\/h3>\n
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.<\/p>\n
# For hourly cronjob,\r\n\r\njob.minute.on(0)\r\njob.hour.during(0, 23)\r\n\r\n# For daily cronjob,\r\n\r\njob.minute.on(0)\r\njob.hour.on(0)\r\n\r\n# For weekly cronjob,\r\n\r\njob.minute.on(0)\r\njob.hour.on(0)\r\njob.dow.on(1)\r\n\r\n# For monthly cronjob,\r\n\r\njob.minute.on(0)\r\njob.hour.on(0)\r\njob.day.on(1)\r\njob.month.during(1, 12)\r\n\r\n# For yearly cronjob,\r\n\r\njob.minute.on(0)\r\njob.hour.on(0)\r\njob.month.on(12)\r\n\r\n# Even if you want to run a cronjob after every reboot,\r\n\r\njob.every_reboot()<\/pre>\n<\/p>\n
Let’s Develop Our Schedule Manager<\/h2>\n
Application Structure<\/h3>\n
.\r\n\u251c\u2500\u2500 app\r\n\u2502\u00a0\u00a0 \u251c\u2500\u2500 __init__.py\r\n\u2502\u00a0\u00a0 \u251c\u2500\u2500 run.py\r\n\u2502\u00a0\u00a0 \u251c\u2500\u2500 scheduler.py\r\n\u2502\u00a0\u00a0 \u2514\u2500\u2500 worker.py\r\n\u251c\u2500\u2500 requirements.txt\r\n\u2514\u2500\u2500 test.txt<\/pre>\nIn our app\/scheduler.py<\/strong> file, we will have the Scheduler <\/strong>class where we will write all the necessary methods to interact with cronjobs.<\/p>\n
The app\/run.py\u00a0<\/strong>file will be used to execute them.<\/p>\n
And lastly, app\/worker.py\u00a0<\/strong>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 <\/strong>file in our case to understand if the cronjob worked or not.<\/p>\n
<\/p>\n
Develop Scheduler Class<\/h3>\n
In our app\/scheduler.py\u00a0<\/strong>file, Let’s import the Crontab module and start writing the Scheduler class.<\/p>\n
\"\"\"\r\n ----------------------------------------------------------------------------\r\n Scheduler module\r\n ----------------------------------------------------------------------------\r\n 1. Can create cron jobs with specific time schedule\r\n\r\n Case Meaning\r\n @reboot Every boot\r\n @hourly 0 * * * *\r\n @daily 0 0 * * *\r\n @weekly 0 0 * * 0\r\n @monthly 0 0 1 * *\r\n @yearly 0 0 1 1 *\r\n ----------------------------------------------------------------------------\r\n 2. can remove all cron jobs\r\n ----------------------------------------------------------------------------\r\n\"\"\"\r\nfrom crontab import CronTab, CronItem\r\n\r\n<\/pre>\nclass Scheduler:\r\n \"\"\"Scheduler class\"\"\"\r\n def __init__(self, command: str):\r\n \"\"\"\r\n Here we will initiate a Cron\r\n and set the command\r\n :param command: command to execute\r\n \"\"\"\r\n self.cron = CronTab(user=True)\r\n self.command = command<\/pre>\n<\/p>\n
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.<\/p>\n
Now let’s write a method to actually create a job.<\/p>\n
def __create_cron_job(self) -> CronItem:\r\n \"\"\"\r\n Create a new job and return it\r\n :return: job\r\n \"\"\"\r\n job = self.cron.new(command=self.command)\r\n return job<\/pre>\nAfter that, we will add methods to execute jobs in many periodic cycles like, minutely, hourly, daily, weekly, monthly, yearly, and after every reboot.<\/p>\n
def minutely_execution(self) -> bool:\r\n \"\"\"\r\n Execute job every minute\r\n :return:\r\n \"\"\"\r\n try:\r\n job = self.__create_cron_job()\r\n job.minute.every(1)\r\n job.enable()\r\n self.cron.write()\r\n if self.cron.render():\r\n return True\r\n return False\r\n except Exception as error:\r\n print(error)\r\n return False\r\n\r\ndef hourly_execution(self) -> bool:\r\n \"\"\"\r\n Execute job every hour\r\n :return:\r\n \"\"\"\r\n try:\r\n job = self.__create_cron_job()\r\n job.minute.on(0)\r\n job.hour.during(0, 23)\r\n job.enable()\r\n self.cron.write()\r\n if self.cron.render():\r\n return True\r\n return False\r\n except Exception as error:\r\n print(error)\r\n return False\r\n\r\ndef daily_execution(self) -> bool:\r\n \"\"\"\r\n Execute job every day\r\n :return:\r\n \"\"\"\r\n try:\r\n job = self.__create_cron_job()\r\n job.minute.on(0)\r\n job.hour.on(0)\r\n job.enable()\r\n self.cron.write()\r\n if self.cron.render():\r\n return True\r\n return False\r\n except Exception as error:\r\n print(error)\r\n return False\r\n\r\ndef weekly_execution(self) -> bool:\r\n \"\"\"\r\n Execute job every week\r\n :return:\r\n \"\"\"\r\n try:\r\n job = self.__create_cron_job()\r\n job.minute.on(0)\r\n job.hour.on(0)\r\n job.dow.on(1)\r\n job.enable()\r\n self.cron.write()\r\n if self.cron.render():\r\n return True\r\n return False\r\n except Exception as error:\r\n print(error)\r\n return False\r\n\r\ndef monthly_execution(self) -> bool:\r\n \"\"\"\r\n Execute job every month\r\n :return:\r\n \"\"\"\r\n try:\r\n job = self.__create_cron_job()\r\n job.minute.on(0)\r\n job.hour.on(0)\r\n job.day.on(1)\r\n job.month.during(1, 12)\r\n job.enable()\r\n self.cron.write()\r\n if self.cron.render():\r\n return True\r\n return False\r\n except Exception as error:\r\n print(error)\r\n return False\r\n\r\ndef yearly_execution(self) -> bool:\r\n \"\"\"\r\n Execute job every year\r\n :return:\r\n \"\"\"\r\n try:\r\n job = self.__create_cron_job()\r\n job.minute.on(0)\r\n job.hour.on(0)\r\n job.month.on(12)\r\n job.enable()\r\n self.cron.write()\r\n if self.cron.render():\r\n return True\r\n return False\r\n except Exception as error:\r\n print(error)\r\n return False\r\n\r\ndef execute_after_reboot(self) -> bool:\r\n \"\"\"\r\n Execute job after every reboot\r\n :return:\r\n \"\"\"\r\n try:\r\n job = self.__create_cron_job()\r\n job.every_reboot()\r\n job.enable()\r\n self.cron.write()\r\n if self.cron.render():\r\n return True\r\n return False\r\n except Exception as error:\r\n print(error)\r\n return False<\/pre>\nWe will also write a method to destroy all cronjobs in case we mess things up.<\/p>\n
def delete_all_jobs(self) -> bool:\r\n \"\"\"\r\n remove all cron jobs\r\n :return:\r\n \"\"\"\r\n try:\r\n self.cron.remove_all()\r\n return True\r\n except Exception as error:\r\n print(error)\r\n return False<\/pre>\nThe Worker In Our Cron Job<\/h3>\n
Well just test our cronjob we will add some file writing code in our app\/worker.py\u00a0<\/strong>file. Which is responsible for writing in the test.txt<\/strong> file.<\/p>\n
from datetime import datetime\r\n\r\n\r\ndef worker(file_path: str):\r\n \"\"\"\r\n This worker could be anything. Any work you desire to be executed\r\n :param file_path: path of the file we wanna write\r\n \"\"\"\r\n with open(file_path, \"a\") as file:\r\n file.write('\\nWorker1 executed at ' + str(datetime.now()))\r\n file.close()\r\n\r\n\r\nif __name__ == \"__main__\":\r\n path = '\/home\/your-location\/' \\\r\n 'schedule_manager\/test.txt'\r\n\r\n worker(path)<\/pre>\n<\/p>\n
Running our program<\/h3>\n