Motivation¶

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, Esmerald comes with a built-in scheduler, saving you tons of headaches and simplifying the process of creating them.

Requirements¶

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

$ pip install esmerald[schedulers]

AsynczConfig¶

The AsynczConfig is the main object that manages the internal scheduler of Esmerald 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 Esmerald is an ASGI framework, it is already provided a default scheduler class that works alongside with the application, the AsyncIOScheduler.

Note

This is for representation and explanation purposes as the EsmeraldScheduler cannot be instantiated, instead, expects parameters being sent upon creating an Esmerald application.

from esmerald import Esmerald
from esmerald.contrib.schedulers.asyncz.config import AsynczConfig

app = Esmerald(scheduler_config=AsynczConfig())

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 Esmerald.

AsynczConfig and the application¶

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

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

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

Enabling the scheduler¶

In order to make sure it does not always start, Esmerald is expecting a flag enable_scheduler to be True. Without the enable_scheduler = True, the scheduler will not start.

The default behaviour is enable_scheduler = False.

from esmerald import Esmerald

app = Esmerald(routes=[...], enable_scheduler=True)

Enabling the scheduler via settings¶

As mentioned in this documentation, Esmerald is unique with settings and therefore the enable_scheduler can also be set to True/False there.

from esmerald import Esmerald, EsmeraldAPISettings


class AppSettings(EsmeraldAPISettings):

    # default is False
    enable_scheduler: bool = True


app = Esmerald(routes=[...])

ESMERALD_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.

Tip

Read more about how to take advantage of the settings and how to use them to leverage your application.

Tasks¶

Tasks are simple pieces of functionality that contains the logic needed to run on a specific time. Esmerald 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 Esmerald 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 esmerald.contrib.schedulers.asyncz.decorator import scheduler

logging.basicConfig()
logging.getLogger("esmerald").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 esmerald import Esmerald
from esmerald.contrib.schedulers.asyncz.config import AsynczConfig

app = Esmerald(
    routes=[...],
    enable_scheduler=True,
    scheduler_config=AsynczConfig(
        tasks={
            "collect_market_data": "accounts.tasks",
            "send_newsletter": "accounts.tasks",
        },
    ),
)

Or from the settings file:

from esmerald import Esmerald, EsmeraldAPISettings
from esmerald.contrib.schedulers.asyncz.config import AsynczConfig


class AppSettings(EsmeraldAPISettings):
    enable_scheduler: bool = True

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


app = Esmerald(routes=[...])

Start the server with the newly created settings.

ESMERALD_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.

Warning

Remember to activate the scheduler by enabling the enable_scheduler flag or else the scheduler will not start.