Running a scheduled task with Nitric in Python

Setting up a scheduled job is a common requirement for many modern applications.

With the Nitric SDK, this task is simplified, and you can even test your scheduled jobs offline without needing Terraform.

In this tutorial, we'll walk through setting up a scheduled job that aggregates data every three days using the Nitric SDK in Python.

If you haven't used the Nitric SDK before, then start with this tutorial.

Step 1: Import Necessary Libraries

Start by importing the necessary libraries.

from nitric.resources import schedule
from nitric.application import Nitric
from nitric.context import IntervalContext

Step 2: Define Your Scheduled Job

Next, define a function for your scheduled job. In this example, we're creating a new scheduled job named "aggregate-data" that will run every three days.

Inside the function, we've added a simple print statement "aggregating data" to demonstrate the job's activity.

from nitric.resources import schedule
from nitric.application import Nitric
from nitric.context import IntervalContext

report_schedule = schedule('run-a-report')

@report_schedule.every('1 days')
async def process_transactions(ctx: IntervalContext):
  # do some processing
  fmt.Println("aggregating data")

Nitric.run()

Test Your Scheduled Job Offline

nitric start

One of the advantages of using the Nitric SDK is the ability to test your scheduled jobs offline. This is particularly useful for debugging and ensuring your jobs are working as expected before deploying them to the cloud.

In just a few steps, you've now created a scheduled job that aggregates data every three days, here are a few more examples showing you how to set up other frequencies.

description	example schedule
Every day	@schedule("work").every("day")
Every 14 hours	@schedule("work").every("14 hours")
Every 30 minutes	@schedule("work").every("30 minutes")
Every day (cron)	@schedule("work").cron("0 0 * * *")