Skip to content

Scheduler

Scheduler wraps APScheduler's AsyncIOScheduler to let you run recurring tasks inside a bot. Jobs are defined with standard cron expressions and are automatically started alongside the bot.

from matrix import Bot

bot = Bot()

@bot.scheduler.job("0 9 * * 1-5")  # weekdays at 09:00
async def morning_standup():
    room = bot.get_room("!abc123:matrix.org")
    await room.send_text("Good morning, team! 🌅")

matrix.scheduler.Scheduler

Scheduler()

The Scheduler class used to schedule tasks for a bot.

Uses APScheduler under the hood.

Source code in matrix/scheduler.py
10
11
12
13
14
15
def __init__(self) -> None:
    """The Scheduler class used to schedule tasks for a bot.

    Uses APScheduler under the hood.
    """
    self.scheduler = AsyncIOScheduler()

scheduler instance-attribute

scheduler = AsyncIOScheduler()

jobs property

jobs

schedule

schedule(cron, func)

Schedule a coroutine function to be run at specified intervals.

:param cron: The cron string defining the schedule. :type cron: str :param func: The coroutine function to run. :type func: Callback

Source code in matrix/scheduler.py
41
42
43
44
45
46
47
48
49
50
51
def schedule(self, cron: str, func: Callback) -> None:
    """
    Schedule a coroutine function to be run at specified intervals.

    :param cron: The cron string defining the schedule.
    :type cron: str
    :param func: The coroutine function to run.
    :type func: Callback
    """
    cron_trigger = CronTrigger(**self._parse_cron(cron))
    self.scheduler.add_job(func, trigger=cron_trigger, name=func.__name__)

start

start()

Start the scheduler.

Source code in matrix/scheduler.py
53
54
55
def start(self) -> None:
    """Start the scheduler."""
    self.scheduler.start()