Task scheduling
Task scheduling allows you to run specific code at set times, regular intervals, or once after a delay. In Linux, tools like cron typically handle this function.
Nestipy offers the nestipy_schedule package that utilize the apscheduler Python package.
Installation
We need to registerScheduleModule to make it work
Python
We can now use from nestipy.common import Module
from app_controller import AppController
from app_service import AppService
from nestipy_schedule import ScheduleModule, ScheduleOption
@Module(
imports=[
ScheduleModule.for_root(ScheduleOption())
],
controllers=[AppController],
providers=[AppService]
)
class AppModule:
...
@Cron, @Interaval or @Timeout decorators inside any controller and provider. ScheduleModule automatically detects and registers these callbacks..
Python
By using from typing import Annotated
from nestipy.common import Injectable
from nestipy.ioc import Inject
from nestipy_schedule import Cron, Interval, Timeout, SchedulerRegistry
@Injectable()
class AppService:
registry: Annotated[SchedulerRegistry, Inject()]
@classmethod
async def get(cls):
return "test"
@classmethod
async def post(cls, data: dict):
return "test"
@classmethod
async def put(cls, id_: int, data: dict):
return "test"
@classmethod
async def delete(cls, id_: int):
return "test"
@classmethod
@Cron("0 0 * * *") # every day at 00:00
async def cron(cls):
print("Running every day at 00:00")
@classmethod
@Interval(2)
async def interval(cls):
print("Running every 2 seconds")
@classmethod
@Timeout(5000)
async def timeout(cls):
print("Running after 5 seconds")
SchedulerRegistry, we can add, get, or remove a schedule job dynamically.
Support us
Nestipy is a project released under the MIT license, meaning it's open source and freely available for use and modification. Its development thrives with the generous contributions of these fantastic individuals.