How to schedule a function to run at a given time in python

In this post I am going to explain couple of different approaches in scheduling tasks that run at a specific time in python

In the previous article I discussed different ways to create a cron job task in python, a task that runs on a frequent basis. In this article, I want to talk about scheduling a task that runs at a given time in python.

So what do I mean by scheduling to run at a given time? To give you an example, let's say you have an application that schedules appointments for users. Now let's say you want to send a message (an email or a text message, for instance) to the customer an hour before their appointment as a reminder. What you need in this case is a task that runs exactly one hour before the appointment and sends the message. This task needs to only run once and not frequently, and needs to be created for each appointment programmatically. So let's see how can we create such tasks in python:

Using threading

Threading is a built-in library in python, which is in fact a high le-vel interface for working with multiple threads. Threading has a method called "Timer" which can run a function when a certain amount of time has passed (i.e. after a delay). To run a function at a given time using "Timer" method, you can calculate the time between now and that specific time in seconds, and pass it to the "Timer" method as the delay to run. Keep in mind that we should consider events such as day time saving in the delay calculations. Here is an example to show how you can use "Timer" method to schedule a function to run at a specific time:

 from datetime import datetime, timedelta
 import threading
 from pytz import timezone

 def job(text):
     print(text)

 run_at = "2022-08-04T19:00:00"

 run_time = datetime.strptime(run_at, '%Y-%m-%dT%H:%M:%S')
 delay = (run_time.astimezone(timezone('UTC')).replace(tzinfo=None) - datetime.utcnow()).total_seconds()
 threading.Timer(delay, job, args=['Task is running now!']).start()
               

In the example above, the "job" function is being scheduled to run at 2022-08-04 19:00:00 local time. Note that the whole utc time zone transformation above is because of the difference in delay that can be caused by day time saving if we use the local time.

Using APScheduler

APScheduler is a third-party library for scheduling tasks, as I mentioned in the previous article (https://crondock.com/how-to-create-cron-job-in-python.html). apscheduler can schedule functions to run at a specific time, as well as on a frequent basis. To install apscheduler, you can run:

pip install apscheduler
               

To schedule a function to run at a specific time using apscheduler, you need to pass "date" as the trigger and also pass the given time you want to run the function as "run_date" to scheduler "add_job". you can also set the "run_at" time zone by passing it to the "timezone" parameter. In the example below, I am using "BackgroundScheduler" as our scheduler and running the "job" function at 2022-08-04 18:40:00 Pacific time:

import time
from apscheduler.schedulers.background import BackgroundScheduler
from datetime import datetime


def job(text):
    print(text)

run_at = "2022-08-04T18:40:00"

scheduler = BackgroundScheduler()
run_time = datetime.strptime(run_at, '%Y-%m-%dT%H:%M:%S')
scheduler.add_job(job, 'date', run_date=run_time, timezone ='US/Pacific', args=['Task is running now!'])

scheduler.start()

while True:
    time.sleep(1)
              

In most cases, you don't need to host your own scheduler system. Hosting your own scheduler system could be costly, since you need a server up and running all the time. Besides, you need to maintain the server and the scheduler application and you may need to scale the server up and down from time to time based on the number of jobs you need to run at a specific time. To avoid that, you can use cloud based and serverless services to host your scheduler. I am going to explain a couple of options you have to run your jobs on cloud based and serverless systems below.

AWS

If you are already a user of AWS, this option could be good for you, however it still could be hard to maintain and complicated for your use cases. Basically, you can schedule "Rule"s in AWS and those rules can run "Lambda function"s. So in this case, what you need to do first is to create an AWS lambda function that contains the code you want to run at the given time. Then in your code you can create a rule and schedule it to trigger at the given time and then connect it to your lambda function, like this:

 import boto3

 events_client = boto3.client('events')
 lambda_fn_arn = "your-lambda-function-arn"
 cron = "30 19 04 08 ? 2022"
 rule_name = "my_rule"

 rule_res = events_client.put_rule(
   Name=rule_name,
   ScheduleExpression='cron({})'.format(cron),
   State='ENABLED',
 )
 events_client.put_targets(
   Rule=rule_name,
   Targets=[
     {
       'Id': "some_id",
       'Arn': lambda_fn_arn,
     },
   ]
 )
               

The above code creates a new "Rule" with a "ScheduleExpression" on AWS. That expression in fact is a cron type expression that triggers only at 19:30 August 4th 2022 UTC. Then the code creates a "Target" that connects that "Rule" to the lambda function you have created.

CronDock

CronDock is a product I built, and in short it is a "container native cloud based cron and workflow" service. It is a serverless framework for running containerized code on demand, based on a schedule or at a specific time. For this example, you are going to need a docker hub account and you use it to host your containerized code in this case. Installing docker cli is also recommended, since it makes it easier to update and push a new version of your container to the docker hub.

So let's say you want to run the following code at a specific time:

 import requests
 import sys

 result = requests.get(sys.argv[1])
 sys.stdout.write(result.text)

               

And let's assume you have this code in a file called "example.py". First, you have to containerize this code using docker. In order to do that, you need to add a file called "Dockerfile" in the same directory you stored "example.py" and add the following commands to it:

FROM python:3
RUN pip install requests
COPY example.py ./
ENTRYPOINT ["python" ,"example.py"]
              

Above block basically says to the docker to create a container image that has python 3 installed on it, then "pip install" the "requests" library on the image and then copies the "example.py" file to the container. Finally, it specifies that when this image is called, it should run "python example.py".

To containerize this code using docker cli, you first need to login to your docker hub account. You can login to your docker account by running the following command (replace yourusername with your docker hub username and yourpassword with your docker hub password):

docker login --username yourusername --password yourpassword
               

Next, you have to run the following command to containerize the code (don't forget to replace yourusername):

docker build -t yourusername/example:latest -f Dockerfile .
               

Above command basically creates a containerized image of the code based on the instruction provided in the "Dockerfile". To run the image locally (for testing), you can use the following command (replace yourusername):

docker run --rm yourusername/example https://crondock.com/
               

Finally, you can push this image to the docker hub by running the following command (replace yourusername):

docker push yourusername/example
               

Now the containerized image of the code lives on the docker hub and CronDock can have access to it. The next step is creating a task to run this image at the specified time. For that, you need a CronDock account, and you can create that here.

CronDock will provide you with an "API key" after signing up, which you can use to access CronDock API. Now you can create a task by simply calling CronDock API using your API key. In python, you can create a CronDock task that runs the above image at "22:30 August 11th 2022 UTC" for instance, by running below code (replace XXXXX with your API key and yourusername with your docker hub username):

  import requests

  headers = {
      "Content-Type": "application/json",
      "Authorization": "Api-Key XXXXX"
  }

  data = {
      "name": "example-at",
      "image": "yourusername/example",
      "type": "at",
      "args": ["https://crondock.com/"],
  	"run_at": "2022-08-11T22:30:00"
    }

  result = requests.post('https://api.crondock.com/task/', headers=headers, json=data)
  print(result.text)
              

And running this code, you see an output like this:

{
	"status": "OK",
	"data": {
		"id": 33,
		"created_at": "2022-08-11T20:39:38.430700Z",
		"name": "example-at-sii0wu",
		"image": "yourusername/example",
		"args": ["https://crondock.com/"],
		"type": "at",
		"schedule": "",
		"run_at": "2022-08-11T22:30:00Z",
		"user": 2,
		"workflow_task": null,
		"workflow_result": null,
		"secret": null
	}
}

              

In the above code, we are using "requests" library to submit a "post" request to CronDock API end point at "https://api.crondock.com/" to create a "task" with type set as "at" (this task type tells the CronDock API that we want to run this task at a specific time). In this task, we specify that we want CronDock to pull and run the "yourusername/example" image and pass the "https://crondock.com/" as the input argument to the image. And we are setting the "run_at" to "2022-08-11T22:30:00", meaning we want CronDock to run this task at "22:30 August 11th 2022 UTC" (CronDock only works with the UTC timezone at the moment).

Sometimes, you need to run multiple images with some dependencies. For instance, you may have to run a specific code in case the first one fails. In such cases, you can use CronDock's workflow, which I explained in more details here: https://crondock.com/how-to-create-a-workflow-programmatically.html

Using CronDock you can also create cron jobs that run on a frequent basis, which is explained here: https://crondock.com/how-to-create-cron-job-in-python.html

Please let me know if you have any question at support@crondock.com

I also have a good document which provides more detailed information on all the requests you can make to CronDock API.