Skip to main content

Task Decorator

To orchestrate a function, you must decorate it with the @task() decorator.

The source code for @task() can be found here.

Import

from magniv.core import task

Fields

Attribute NameTypeRequired
schedulestrYes
descriptionstrNo
keystrNo
resourcesdict[str, str]No
webhook_triggerboolNo
on_successlistNo

Usage examples

from magniv.core import task

@task(schedule="@daily", description="Daily hello world!", key="hello-world")
def hello_world():
print("hello world")

@task(schedule="@daily", resources={"cpu": "500m", "memory": "2Gi"})
def resourceful():
print("i'm using custom resources")

@task(schedule="@daily", webhook_trigger=True)
def captain_hook():
print("i can be triggered via webhook, yarrrr")

@task(schedule="@daily", on_success=["hello_world"])
def trigger_hello_world():
print("i'm about to greet the world!")