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

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.

Example

@bot.schedule("0 9 * * 1-5")
async def morning_update() -> None:
    await room.send("Good morning!")

Source code in matrix/scheduler.py

def schedule(self, cron: str, func: Callback) -> None:
    """
    Schedule a coroutine function to be run at specified intervals.

    ## Example

    ```python
    @bot.schedule("0 9 * * 1-5")
    async def morning_update() -> None:
        await room.send("Good morning!")
    ```
    """
    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

def start(self) -> None:
    """Start the scheduler."""
    self.scheduler.start()