​
Creating Cron tasks

Cron tasks run repeatedly on a specified cron schedule. To create a Cron task, use tilebox.workflows.recurrent_tasks.CronTask as your tasks base class instead of the regular tilebox.workflows.Task.

Python
from tilebox.workflows import ExecutionContext
from tilebox.workflows.recurrent_tasks import CronTask

class MyCronTask(CronTask):
    message: str

    def execute(self, context: ExecutionContext) -> None:
        print(f"Hello {self.message} from a Cron Task!")
        # self.trigger is an attribute of the CronTask class,
        # which contains information about the trigger event
        # that caused this task to be submitted as part of a job
        print(f"This task was triggered at {self.trigger.time}")

​
Registering a Cron trigger

After implementing a Cron task, register it to be triggered according to a Cron schedule. When the Cron expression matches, a new job is submitted consisting of a single task instance derived from the Cron task prototype.

Python
from tilebox.workflows import Client

client = Client()
recurrent_tasks = client.recurrent_tasks()
cron_task = recurrent_tasks.create_recurring_cron_task(
    "my-cron-task",  # name of the recurring cron task
    "dev-cluster",  # cluster slug to submit jobs to
    MyCronTask(message="World"),  # the task (and its input parameters) to run repeatedly
    cron_triggers=[
        "12 * * * *",  # run every hour at minute 12
        "45 18 * * *",  # run every day at 18:45
        "30 13 * * 3",  # run every Wednesday at 13:30
    ],
)

The syntax for specifying the cron triggers is a CRON expression. A helpful tool to test your cron expressions is crontab.guru.

​
Starting a Cron Task Runner

With the Cron task registered, a job is submitted whenever the Cron expression matches. But unless a task runner is available to execute the Cron task the submitted jobs remain in a task queue. Once an eligible task runner becomes available, all jobs in the queue are executed.

Python
from tilebox.workflows import Client

client = Client()
runner = client.runner("dev-cluster", tasks=[MyCronTask])
runner.run_forever()

If this task runner runs continuously, its output may resemble the following:

Output
Hello World from a Cron Task!
This task was triggered at 2023-09-25 16:12:00
Hello World from a Cron Task!
This task was triggered at 2023-09-25 17:12:00
Hello World from a Cron Task!
This task was triggered at 2023-09-25 18:12:00
Hello World from a Cron Task!
This task was triggered at 2023-09-25 18:45:00
Hello World from a Cron Task!
This task was triggered at 2023-09-25 19:12:00

​
Inspecting in the Console

The Tilebox Console provides a straightforward way to inspect all registered Cron tasks.

Use the console to view, edit, and delete the registered Cron tasks.

​
Deleting Cron triggers

To delete a registered Cron task, use recurrent_tasks.delete. After deletion, no new jobs will be submitted by that Cron trigger. Past jobs already triggered will still remain queued.

Python
from tilebox.workflows import Client

client = Client()
recurrent_tasks = client.recurrent_tasks()

# delete the task as returned by create_recurring_cron_task
recurrent_tasks.delete(cron_task)

# or manually by id:
recurrent_tasks.delete("0190bafc-b3b8-88c4-008b-a5db044380d0")

​
Submitting Cron jobs manually

You can submit Cron tasks as regular tasks for testing purposes or as part of a larger workflow. To do so, instantiate the task with a specific trigger time using the once method.

Submitting a job with a Cron task using once immediately schedules the task, and a runner may pick it up and execute it. The trigger time set in the once method does not influence the execution time; it only sets the self.trigger.time attribute for the Cron task.

Python
from datetime import datetime, timezone

job_client = client.jobs()

# create a Cron task prototype
task = MyCronTask(message="Hello")

# submitting it directly won't work: raises ValueError:
# CronTask cannot be submitted without being triggered. Use task.once().
job_client.submit("manual-cron-job", task, cluster="dev-cluster")

# specify a trigger time to submit the task as a regular task
triggered_task = task.once()  # same as task.once(datetime.now())
job_client.submit("manual-cron-job", triggered_task, cluster="dev-cluster")

# simulate a trigger at a specific time
triggered_task = task.once(datetime(2030, 12, 12, 15, 15, tzinfo=timezone.utc))
# the task will be scheduled to run immediately, even with a future trigger time
# but the self.trigger.time will be 2023-12-12T15:15:00Z for the task instance
job_client.submit("manual-cron-job", triggered_task, cluster="dev-cluster")

Was this page helpful?

Suggest editsRaise issue
Recurrent TasksStorage Event Triggers