`Scheduler` decorator¶

Almost every application in one way or another needs some sort of automated scheduler to run automated tasks. In that in mind and with the help of the great widely used Asyncz, Lilya comes with a built-in scheduler, saving you tons of headaches and simplifying the process of creating them.

Requirements¶

Lilya uses asyncz for this integration. You can install by running:

$ pip install asyncz

AsynczConfig¶

The AsynczConfig is the main object that manages the internal scheduler of Lilya with asyncz expecting:

scheduler_class - An instance of the Asyncz schedule type. Passed via scheduler_class.

^{Default: AsyncIOScheduler}
tasks - A python dictionary of both key, value string mapping the tasks. Passed via scheduler_tasks.

^{Default: {}}
timezone - The timezone of the scheduler. Passed via timezone.

^{Default: UTC}
configurations - A python dictionary containing some extra configurations for the scheduler. Passed via scheduler_configurations.
kwargs - Any keyword argument that can be passed and injected into the scheduler_class.

Since Lilya is an ASGI framework, it is already provided a default scheduler class that works alongside with the application, the AsyncIOScheduler.

from lilya.apps import Lilya
from lilya.contrib.schedulers.asyncz.config import AsynczConfig

scheduler_config = AsynczConfig()
app = Lilya(on_startup=[scheduler_config.start], on_shutdown=[scheduler_config.shutdown])

You can have your own scheduler config class as well, check the SchedulerConfig. for more information.

Warning

Anything else that does not work with AsyncIO is very likely also not to work with Lilya.

AsynczConfig and the application¶

This is the default Lilya integration with Asyncz and the class can be accessed via:

from lilya.contrib.schedulers.asyncz.config import AsynczConfig

Because this is an Lilya offer, you can always implement your own version if you don't like the way Lilya handles the Asyncz default integration and adapt to your own needs. This is thanks to the SchedulerConfig from where AsynczConfig is derived.

Tasks¶

Tasks are simple pieces of functionality that contains the logic needed to run on a specific time. Lilya does not enforce any specific file name where the tasks should be, you can place them anywhere you want.

Once the tasks are created, you need to pass that same information to your Lilya instance.

Tip

There are more details about how to create tasks in the next section.

accounts/tasks.py

import logging

from loguru import logger

from asyncz.triggers import IntervalTrigger
from lilya.contrib.schedulers.asyncz.decorator import scheduler

logging.basicConfig()
logging.getLogger("lilya").setLevel(logging.DEBUG)


@scheduler(name="collect_data", trigger=IntervalTrigger(hours=12), max_instances=3)
def collect_market_data():
    logger.error("Collecting market data")
    ...


@scheduler(
    name="send_newsletter",
    trigger=IntervalTrigger(days=7),
    max_instances=3,
)
def send_newsletter():
    logger.warning("sending email newsletter!")
    ...

There are two tasks created, the collect_market_data and send_newsletter which are placed inside a accounts/tasks.

Now it is time to tell the application to activate the scheduler and run the tasks based on the settings provided into the scheduler handler.

from lilya.apps import Lilya
from lilya.contrib.schedulers.asyncz.config import AsynczConfig

scheduler_config=AsynczConfig(
    tasks={
        "collect_market_data": "accounts.tasks",
        "send_newsletter": "accounts.tasks",
    },
),

app = Lilya(
    routes=[...],
    on_startup=[scheduler_config.start],
    on_shutdown=[scheduler_config.shutdown],
)

Or from the settings file:

from lilya.apps import Lilya
from lilya.conf.global_settings import Settings
from lilya.conf import settings
from esmerald.contrib.schedulers.asyncz.config import AsynczConfig


# This is an example of how to configure tasks using the Asyncz scheduler in Lilya.
class AppSettings(Settings):
    enable_scheduler: bool = True

    @property
    def scheduler_config(self) -> AsynczConfig:
        return AsynczConfig(
            tasks={
                "collect_market_data": "accounts.tasks",
                "send_newsletter": "accounts.tasks",
            },
            stores=...,
            executors=...,
        )


# In theory, this is in a different file, but for the sake of this example, we are defining it here.
app = Lilya(
    routes=[...],
    on_startup=[settings.scheduler_config.start],
    on_shutdown=[settings.scheduler_config.shutdown],
)

Start the server with the newly created settings.

MacOS & LinuxWindows

LILYA_SETTINGS_MODULE=AppSettings uvicorn src:app --reload

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [28720]
INFO:     Started server process [28722]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

$env:LILYA_SETTINGS_MODULE="AppSettings"; uvicorn src:app --reload

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [28720]
INFO:     Started server process [28722]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

The scheduler_tasks is expecting a python dictionary where the both key and value are strings.

key - The name of the task.
value - Where the task is located.

Scheduler decorator¶