Python: Sched for Automating Tasks in Python
There are many python underdog libraries and scheduling is one of the them.
One of the most useful features of Python is the ability to schedule tasks to run at specific times or on a regular basis. For that Python provides a built-in module called sched. With the sched module, you can automate repetitive tasks, schedule reminders, or run scripts at specific times.
In this article, I will explore the concept of scheduling in Python and will also go through couple of examples on how to implement it.
Getting Started:
A scheduler is a tool that allows you to run a specific task at a specified time. This can be useful for a variety of applications, such as automating the execution of scripts, sending emails, or performing backups. In Python, the “sched” library provides a simple and easy-to-use interface for scheduling tasks.
The first step in using the sched module is to create an instance of the sched.scheduler class. This class is used to schedule and execute tasks, and provides several methods for adding, removing, and managing scheduled events.
import sched
scheduler = sched.scheduler()In this example, I have created a new sched.scheduler object called scheduler.
Once we have a sched instance, we can use it to schedule tasks by calling the scheduler.enter method. This method takes several arguments, including the time at which the task should be executed, the priority of the task, and a function to be executed.
Here’s an example of how to schedule a simple task using the scheduler.enter method:
import sched
import time
def say_hello():
print("Hello, world!")
scheduler = sched.scheduler(time.time, time.sleep)
scheduler.enter(5, 1, say_hello, ())
scheduler.run()In this example, I have defined a function called say_hello that prints "Hello, world!" to the console. Then created a new sched.scheduler object and passed in the time.time and time.sleep functions as arguments. These functions are used to calculate the time at which the task should be executed and to delay the program until that time.
Then called the scheduler.enter method, passing in the time at which the task should be executed (5 seconds from the current time), a priority of 1, the say_hello function to be executed, and an empty tuple of arguments.
Finally, called the scheduler.run method to start the scheduler and execute the scheduled task. The program will pause for 5 seconds before executing the say_hello function.
Lets Understand how it Works
The sched module in Python uses a priority queue to keep track of scheduled events. When we create a scheduler instance, it initializes an empty priority queue.
When we use the enter method to schedule an event, the scheduler adds the event to the priority queue. The event is added with a specific execution time and priority, which determines its position in the queue.
The scheduler then waits until the first event in the queue is due for execution. It does this by calling the sleep function, which blocks the program until the next event is due.
When an event is due for execution, the scheduler retrieves it from the priority queue and executes the function associated with the event. The scheduler then repeats this process for all events in the queue until there are no more events left.
If new events are added to the scheduler while it is running, they are added to the priority queue in order of their scheduled execution time. This ensures that events are executed in the correct order.
The scheduler also provides a way to cancel scheduled events using the cancel method. This method takes the event's reference as an argument and removes it from the priority queue if it is still pending.
Lets look at some of the implementation of sched in Python Program
Running Tasks Repeatedly:
In addition to scheduling tasks to run at specific times, the sched library also allows you to schedule tasks to run repeatedly.
To do this, you can use the enter or enterabs method in conjunction with a loop.
Here’s an example of how to schedule a task to run every 5 seconds:
import sched
import time
def say_hello():
print("Hello, world!")
scheduler = sched.scheduler(time.time, time.sleep)
def repeat_task():
scheduler.enter(5, 1, say_hello, ())
scheduler.enter(5, 1, repeat_task, ())
repeat_task()
scheduler.run()In this example, I have defined a new function called repeat_task that uses the scheduler.enter method to schedule the say_hello function to be executed every 5 seconds. The repeat_task function then calls itself to reschedule the say_hello function again after 5 seconds.
Then it will call the repeat_task function to start the repeating task, followed by the scheduler.run method to execute the scheduled events
Using `enterabs` to Schedule a task to run at a specific time:
In addition to scheduling tasks to run after a delay or at a repeating interval, the sched module also allows you to schedule tasks to run at a specific time. You can use the scheduler.enterabs method to do this.
Here’s an example of how to schedule a task to run at a specific time:
import sched
import time
def say_hello():
print("Hello, world!")
scheduler = sched.scheduler(time.time, time.sleep)
# Schedule the task to run at a specific time
specific_time = time.time() + 5 # 5 seconds from now
scheduler.enterabs(specific_time, 1, say_hello, ())
scheduler.run()In this example, I have defined a function called say_hello that prints "Hello, world!" to the console. Then created a new sched.scheduler object and pass in the time.time and time.sleep functions as arguments.
Then used the time.time() function to calculate a specific time 5 seconds from the current time, and pass this time to the scheduler.enterabs method along with a priority of 1, the say_hello function to be executed, and an empty tuple of arguments.
Finally, called the scheduler.run method to start the scheduler and execute the scheduled task. The program will pause until the specified time, at which point the say_hello function will be executed.
Scheduling Multiple Tasks:
In addition to scheduling individual tasks, you can also use the sched module to schedule multiple tasks. To do this, you can call the enter or enterabs methods multiple times with different task information.
Here’s an example of how to schedule two tasks to run at different intervals:
import sched
import time
def task_one():
print("Task One - Hello, world!")
def task_two():
print("Task Two - Hello, world!")
scheduler = sched.scheduler(time.time, time.sleep)
# Schedule task one to run after 2 seconds
scheduler.enter(2, 1, task_one, ())
# Schedule task two to run after 5 seconds
scheduler.enter(5, 1, task_two, ())
scheduler.run()In this example, I have defined two functions, task_one and task_two, that both print "Hello, world!" to the console. Then created a new sched.scheduler object and pass in the time.time and time.sleep functions as arguments.
Then used the scheduler.enter method to schedule task_one to run after 2 seconds with a priority of 1, and task_two to run after 5 seconds with the same priority.
Finally, called the scheduler.run method to start the scheduler and execute the scheduled tasks. The program will pause for 2 seconds, execute task_one, pause for another 3 seconds, and then execute task_two.
Scheduling Tasks with different Priorities:
When scheduling tasks with the sched module, you can also assign different priorities to each task. Tasks with a lower priority number will be executed before tasks with a higher priority number.
Here’s an example of how to schedule two tasks with different priorities:
import sched
import time
def task_one():
print("Task One - Hello, world!")
def task_two():
print("Task Two - Hello, world!")
scheduler = sched.scheduler(time.time, time.sleep)
# Schedule task one with priority 1
scheduler.enter(2, 1, task_one, ())
# Schedule task two with priority 2
scheduler.enter(5, 2, task_two, ())
scheduler.run()In this example, I defined two functions, task_one and task_two, that both print "Hello, world!" to the console. Then created a new sched.scheduler object and pass in the time.time and time.sleep functions as arguments.
Then used the scheduler.enter method to schedule task_one to run after 2 seconds with a priority of 1, and task_two to run after 5 seconds with a priority of 2.
Finally, I called the scheduler.run method to start the scheduler and execute the scheduled tasks. The program will pause for 2 seconds, execute task_one, pause for another 3 seconds, and then execute task_two.
Scheduling Tasks with a Cancel Method:
Sometimes it may be necessary to cancel a scheduled task before it is executed. The sched module provides a cancel method that can be used to remove a scheduled task from the scheduler.
Here’s an example of how to schedule a task and then cancel it:
import sched
import time
def task_one():
print("Task One - Hello, world!")
def task_two():
print("Task Two - Hello, world!")
scheduler = sched.scheduler(time.time, time.sleep)
# Schedule task one to run after 2 seconds
task_one_event = scheduler.enter(2, 1, task_one, ())
# Schedule task two to run after 5 seconds
task_two_event = scheduler.enter(5, 1, task_two, ())
# Cancel task one
scheduler.cancel(task_one_event)
scheduler.run()In this example, I have defined two functions, task_one and task_two, that both print "Hello, world!" to the console. Then I create a new sched.scheduler object and pass in the time.time and time.sleep functions as arguments.
Then used the scheduler.enter method to schedule task_one to run after 2 seconds with a priority of 1, and task_two to run after 5 seconds with the same priority.
Next, I assigned the return value of the scheduler.enter method to the task_one_event and task_two_event variables. This allows us to reference the events later on.
Finally, I am calling the scheduler.cancel method and pass in the task_one_event variable to remove the task_one function from the scheduler. And then called scheduler.run to start the scheduler and execute the remaining scheduled tasks, which in this case is only task_two.
Running Backups:
Scheduling backups is a common task in many applications. In Python, we can use the sched module to schedule backup operations at a specific time or interval.
Here’s an example of how to use the sched module to schedule a backup operation:
import sched
import time
import shutil
def backup_files():
source = '/path/to/source/files'
destination = '/path/to/backup/location'
shutil.copytree(source, destination)
def schedule_backup():
# Create a new scheduler
scheduler = sched.scheduler(time.time, time.sleep)
# Schedule the backup to run at 1:00 AM every day
backup_time = time.strptime('01:00:00', '%H:%M:%S')
backup_event = scheduler.enterabs(time.mktime(backup_time), 1, backup_files, ())
# Start the scheduler
scheduler.run()
schedule_backup()In this example, I have defined a function backup_files that copies a directory of files from a source location to a backup location. Then defined a function schedule_backup that creates a new sched.scheduler object and schedules the backup_files function to run every day at 1:00 AM.
To schedule the backup, first use the time.strptime function to convert the backup time from a string format of HH:MM:SS to a struct_time object. Then use the time.mktime function to convert the struct_time object to a UNIX timestamp, which is the format expected by the sched module.
Next, use the scheduler.enterabs method to schedule the backup_files function to run at the specified time with a priority of 1. Then call scheduler.run to start the scheduler and execute the scheduled backup.
Sending Emails:
In this example, I will show you how to use the sched module to schedule sending an email at a specific time.
import sched
import time
import smtplib
from email.mime.text import MIMEText
def send_email(subject, message, from_addr, to_addr, smtp_server):
# Create the email message
email = MIMEText(message)
email['Subject'] = subject
email['From'] = from_addr
email['To'] = to_addr
# Send the email
with smtplib.SMTP(smtp_server) as server:
server.send_message(email)
def send_scheduled_email(subject, message, from_addr, to_addr, smtp_server, scheduled_time):
# Create the scheduler
scheduler = sched.scheduler(time.time, time.sleep)
# Schedule the email
scheduler.enterabs(scheduled_time, 1, send_email, argument=(subject, message, from_addr, to_addr, smtp_server))
# Start the scheduler
scheduler.run()
subject = 'Test Email'
message = 'This is a test email'
from_addr = 'test@example.com'
to_addr = 'test@example.com'
smtp_server = 'smtp.test.com'
scheduled_time = time.time() + 60 # Schedule the email to be sent in 1 minute
send_scheduled_email(subject, message, from_addr, to_addr, smtp_server, scheduled_time)In this example, I have defined a function send_scheduled_email that takes the subject, message, sender email address, recipient email address, SMTP server address, and scheduled time as arguments.
Then created a sched.scheduler object, which is used to schedule events. Here I have used the enterabs method of the scheduler to schedule the email to be sent at the specified time. The enterabs method takes a timestamp, a priority level, a function to call, and arguments to pass to the function as arguments.
When the scheduled time arrives, the scheduler calls the send_email function with the specified arguments.
We can then call the send_scheduled_email function with the email details and the scheduled time. In this example, we schedule the email to be sent in 1 minute from the current time.
NOTE:
You can schedule your tasks like calling a different script which performs some other functionalities like perform_etl, sql_load, running_batch_job, dataquality_check, report_generation, etc. with the below format using `subprocess`:
import sched
import time
import subprocess
s = sched.scheduler(time.time, time.sleep)
def your_function():
subprocess.call(['/path/to/your/script.py'])
# Schedule the job to run every Sunday at 2:00 AM
s.enterabs(time.mktime(time.strptime("2023–02–26 02:00:00", "%Y-%m-%d %H:%M:%S")), 1, cleaning_job)
# Start the scheduler
s.run()In conclusion, the sched module in Python provides a simple and effective way to schedule tasks to run automatically.
Whether you need to run a task at a specific time or on a regular basis, the sched module has the tools to help you achieve your goals.
I believe this article helped you in getting the basic knowledge of how and where we can use sched.
There are other use cases of sched as I am not sure how extensively this is being used but yes, This scheduler runs in the background and manages a queue of events, executing them when their scheduled time arrives so you don’t have to worry about using separate threads or processes.
Connect with me on LinkedIn
