Table of Contents

Schedule periodic tasks inside your Pyronaut applications

Learn how to schedule periodic tasks inside your Micronaut microservices.

1. Getting Started

In this guide, we will create a Python application built with Pyronaut.

Nowadays it is pretty usual to have some kind of cron or scheduled task that needs to run every midnight, every hour, a few times a week,…​

In this guide you will learn how to use Micronaut capabilities to schedule periodic tasks inside a Micronaut microservice.

2. What you will need

To complete this guide, you will need the following:

  • Some time on your hands

  • GraalPy installed and the Pyronaut CLI available locally

3. Solution

We recommend that you follow the instructions in the next sections and create the application step by step. However, you can go right to the completed example.

4. Writing the Application

Create an application using the Pyronaut CLI (Command Line Interface) or Pyronaut Launch

pyronaut create example.micronaut.micronautguide

4.1. Logging

The generated Pyronaut application configures logging in src/main.py and logs at INFO by default.

4.2. Creating a Job

Create the HelloWorldJob class:

src/example/micronaut/hello_world_job.py
import logging
from datetime import datetime

from jakarta.inject import Singleton
from micronaut.scheduling.annotation import Scheduled

LOG = logging.getLogger(__name__)  (2)

def _current_time() -> str:
    now = datetime.now()
    return f"{now.day}/{now.month}/{now.year} {now:%I:%M:%S}"


@Singleton  (1)
class HelloWorldJob:
    @Scheduled(fixedDelay="10s")  (3)
    def execute_every_ten(self) -> None:
        LOG.info("Simple Job every 10 seconds: %s", _current_time())

    @Scheduled(fixedDelay="45s", initialDelay="5s")  (4)
    def execute_every_forty_five(self) -> None:
        LOG.info("Simple Job every 45 seconds: %s", _current_time())
1 Use jakarta.inject.Singleton to designate a class as a singleton.
2 Inject a Logger.
3 Create trigger every 10 seconds
4 Create another trigger every 45 seconds with an initial delay of 5 seconds (5000 millis)

After a few seconds, you will see the following output:

... Simple Job every 10 seconds :15/5/2018 12:48:02 (1)
... Simple Job every 45 seconds :15/5/2018 12:48:07 (2)
... Simple Job every 10 seconds :15/5/2018 12:48:12 (3)
... Simple Job every 10 seconds :15/5/2018 12:48:22
... Simple Job every 10 seconds :15/5/2018 12:48:32
... Simple Job every 10 seconds :15/5/2018 12:48:42
... Simple Job every 45 seconds :15/5/2018 12:48:52 (4)
... Simple Job every 10 seconds :15/5/2018 12:48:52
1 First execution of 10 seconds Job after the application starts
2 The 45 seconds Job starts 5 seconds after the application starts
3 Second execution of 10 seconds Job 10 seconds after the first execution
4 Second execution of 45 seconds Job 45 seconds after the first execution

4.3. Business logic in dedicated Use Cases

Although the previous example is valid, usually you don’t want to put your business logic in a Job. A better approach is to create an additional bean that the Job invokes. This approach decouples your business logic from the scheduling logic. Moreover, it facilitates testing and maintenance. Let’s see an example:

Create the following use case:

src/example/micronaut/email_use_case.py
import logging
from datetime import datetime

from jakarta.inject import Singleton

LOG = logging.getLogger(__name__)


def _current_time() -> str:
    now = datetime.now()
    return f"{now.day}/{now.month}/{now.year} {now:%I:%M:%S}"


@Singleton
class EmailUseCase:
    def send(self, user: str, message: str) -> None:
        LOG.info("Sending email to %s: %s at %s", user, message, _current_time())

And then the Job:

src/example/micronaut/daily_email_job.py
from jakarta.inject import Singleton
from micronaut.scheduling.annotation import Scheduled

from .email_use_case import EmailUseCase

@Singleton  (1)
class DailyEmailJob:
    def __init__(self, email_use_case: EmailUseCase):  (2)
        self.email_use_case = email_use_case

    @Scheduled(cron="0 30 4 1/1 * ?")  (3)
    def execute(self) -> None:
        self.email_use_case.send("john.doe@micronaut.example", "Test Message")  (4)
1 Use jakarta.inject.Singleton to designate a class as a singleton.
2 Constructor injection.
3 Trigger the Job once a day at 04:30 AM
4 Call the injected use case

4.4. Scheduling a Job Manually

Consider the following scenario. You want to send every user an email two hours after they register on your application and ask them about their experiences during this first interaction.

For this guide, we will schedule a Job to trigger after one minute.

To test it, we will call a new use case named RegisterUseCase twice when the application starts.

Modify Application.py:

src/example/micronaut/application.py
import time

from jakarta.inject import Singleton
from micronaut.runtime.server.event import ServerStartupEvent
from micronaut.runtime.event.annotation import EventListener

from .register_use_case import RegisterUseCase

@Singleton  (1)
class Application:  (2)
    def __init__(self, register_use_case: RegisterUseCase):  (3)
        self.register_use_case = register_use_case

    @EventListener
    def onApplicationEvent(self, event: ServerStartupEvent) -> None:  (4)
        self.register_use_case.register("harry@micronaut.example")
        time.sleep(20)
        self.register_use_case.register("ron@micronaut.example")
1 Use jakarta.inject.Singleton to designate a class as a singleton.
2 Listen to the event ServerStartupEvent
3 Constructor injection of RegisterUseCase
4 onApplicationEvent is invoked when the application starts. Subscribing to an event is as easy as implementing ApplicationEventListener.

Create a runnable task EmailTask.py

src/example/micronaut/email_task.py
from java.lang import Runnable

from .email_use_case import EmailUseCase

class EmailTask(Runnable):
    def __init__(self, email_use_case: EmailUseCase, email: str, message: str):
        self.email_use_case = email_use_case
        self.email = email
        self.message = message

    def run(self) -> None:
        self.email_use_case.send(self.email, self.message)

Create RegisterUseCase.py, which schedules the previous task.

src/example/micronaut/register_use_case.py
import logging
from datetime import datetime
from typing import Annotated

from jakarta.inject import Named, Singleton
from java.time import Duration
from micronaut.scheduling import TaskExecutors, TaskScheduler

from .email_task import EmailTask
from .email_use_case import EmailUseCase

LOG = logging.getLogger(__name__)

def _current_time() -> str:
    now = datetime.now()
    return f"{now.day}/{now.month}/{now.year} {now:%I:%M:%S}"


@Singleton
class RegisterUseCase:
    def __init__(
        self,
        email_use_case: EmailUseCase,  (1)
        task_scheduler: Annotated[TaskScheduler, Named(TaskExecutors.SCHEDULED)],  (2)
    ):
        self.email_use_case = email_use_case
        self.task_scheduler = task_scheduler

    def register(self, email: str) -> None:
        LOG.info("saving %s at %s", email, _current_time())
        self._schedule_followup_email(email, "Welcome to the Micronaut framework")

    def _schedule_followup_email(self, email: str, message: str) -> None:
        task = EmailTask(self.email_use_case, email, message)  (3)
        self.task_scheduler.schedule(Duration.ofMinutes(1), task)  (4)
1 Constructor injection of EmailUseCase
2 Inject the TaskScheduler bean
3 Create a Runnable task
4 Schedule the task to run a minute from now

If you execute the above code, you will see the Job being executed one minute after we schedule it and with the supplied email address.

INFO  example.micronaut.RegisterUseCase - saving harry@micronaut.example at 15/5/2018 06:25:14
INFO  example.micronaut.RegisterUseCase - saving ron@micronaut.example at 15/5/2018 06:25:34
INFO  example.micronaut.EmailUseCase - Sending email to harry@micronaut.example : Welcome to Micronaut at 15/5/2018 06:26:14
INFO  example.micronaut.EmailUseCase - Sending email to ron@micronaut.example : Welcome to the Micronaut framework at 15/5/2018 06:26:34

5. Summary

During this guide, we learned how to configure Jobs using the @Scheduled annotation using fixedDelay, initialDelay, and cron, as well as manually configuring Jobs with a TaskScheduler and tasks.

6. Next Steps

Explore more features with Micronaut Guides.

7. License

All guides are released with an Apache License 2.0 for the code and a Creative Commons Attribution 4.0 license for the writing and media (images).