Task & TaskScheduler

Purpose

A task is a periodic event run on a separate thread and managed by the TaskScheduler.

Tasks.py

Abstract class for creating a task.

Method definitions

Constructor

def __init__(self, schedule_method: Callable):
    self.__schedule_method = schedule_method

The schedule_method is the function that determines when the task needs to run. For example:

import schedule
schedule.every().day.at("00:00").do # This task runs every day at midnight.
schedule.every(5).seconds.do # This task runs every five seconds.

do

This is where all the magic happens. All the code written here will be executed.

async def do(self):

TaskScheduler.py

Registers all tasks and executes them at the right time on a separate thread.

⚠️ All tasks run on the TaskScheduler thread!

Methods definition

add_tasks

def add_tasks(self, tasks: List[Task]):

Adds a list of tasks to the scheduler.

stop

def stop(self):

Stops the TaskScheduler.

⚠️ You must use this method to stop the thread.

How to create a task

By default, all tasks are stored in the directory bot/application/task. Create a new file that will represent your task. Your new task must inherit from the base class like so:

class MyNewlyTask(Task):

Don't forget to call the base constructor with the schedule_method. Lastly, make sure to override the async def do method.

ℹ️ The do method is async, so you can perform await operations.

⚠️ The task is not a thread, so make sure not to perform blocking operations.

How to register a task

To register a new task, go to bot/config/context/development_context.py and bot/config/context/production_context.py. Locate the method _instantiate_tasks. Simply add your new task here. The base class will automatically register and schedule each task.