Triggers¶

Triggers are used to determine when the event should be triggered. It can be based on time, or some other condition. You can create custom triggers by inheriting from BaseTrigger class.

Don't run CPU intensitve or thread-block IO task

AioClock's trigger are all running in async, only on one CPU. So, if you run a CPU intensive task, or a task that blocks the thread, then it will block the entire event loop. If you have a sync IO task, then it's recommended to use run_in_executor to run the task in a separate thread. Or use similiar libraries like asyncer or trio to run the task in a separate thread.

BaseTrigger ¶

Bases: BaseModel, ABC, Generic[TriggerTypeT]

Base class for all triggers. A trigger is a way to determine when the event should be triggered. It can be based on time, or some other condition.

The way trigger are used is as follows

An async function which is a task, is decorated with framework, and trigger is the arguement for the decorator
get_waiting_time_till_next_trigger is called to get the time in seconds, after which the event should be triggered.
If the time is not None, then it logs the time that is predicted for the event to be triggered.
trigger_next is called immidiately after that, which triggers the event.

You can create trigger by yourself, by inheriting from BaseTrigger class.

Example

from aioclock.triggers import BaseTrigger
from typing import Literal

class Forever(BaseTrigger[Literal["Forever"]]):
    type_: Literal["Forever"] = "Forever"

    def should_trigger(self) -> bool:
        return True

    async def trigger_next(self) -> None:
        return None

    async def get_waiting_time_till_next_trigger(self):
        if self.should_trigger():
            return 0
        return None

Attributes:

Name	Type	Description
`type_`	`TriggerTypeT`	Type of the trigger. It is a string, which is used to identify the trigger's name. You can change the type by using `Generic` type when inheriting from `BaseTrigger`.
`expected_trigger_time`	`Union[datetime, None]`	Expected time when the event should be triggered. This gets updated by Task Runner. It can be used on API layer, to know when the event is expected to be triggered.

trigger_next `abstractmethod` `async` ¶

trigger_next() -> None

trigger_next keep track of the event, and triggers the event. The function shall return when the event is triggered and should be executed.

Source code in aioclock/triggers.py

@abstractmethod
async def trigger_next(self) -> None:
    """
    `trigger_next` keep track of the event, and triggers the event.
    The function shall return when the event is triggered and should be executed.
    """

should_trigger ¶

should_trigger() -> bool

should_trigger checks if the event should be triggered or not. If not, then the event will not be triggered anymore. You can save the state of the trigger and task inside the instance, and then check if the event should be triggered or not. For instance, in LoopCounter trigger, it keeps track of the number of times the event has been triggered, and then checks if the event should be triggered or not.

Source code in aioclock/triggers.py

def should_trigger(self) -> bool:
    """
    `should_trigger` checks if the event should be triggered or not.
    If not, then the event will not be triggered anymore.
    You can save the state of the trigger and task inside the instance, and then check if the event should be triggered or not.
    For instance, in `LoopCounter` trigger, it keeps track of the number of times the event has been triggered,
    and then checks if the event should be triggered or not.
    """
    return True

get_waiting_time_till_next_trigger `abstractmethod` `async` ¶

get_waiting_time_till_next_trigger() -> Union[float, None]

Returns the time in seconds, after which the event should be triggered. Returns None, if the event should not trigger anymore.

Source code in aioclock/triggers.py

@abstractmethod
async def get_waiting_time_till_next_trigger(self) -> Union[float, None]:
    """
    Returns the time in seconds, after which the event should be triggered.
    Returns None, if the event should not trigger anymore.
    """
    ...

Forever ¶

Bases: BaseTrigger[Literal[FOREVER]]

A trigger that is always triggered imidiately.

Example

    from aioclock import AioClock, Forever

    app = AioClock()

    # instead of this:
    async def my_task():
        while True:
            try:
                await asyncio.sleep(3)
                1/0
            except DivisionByZero:
                pass

    # use this:
    @app.task(trigger=Forever())
    async def my_task():
        await asyncio.sleep(3)
        1/0

Attributes:

Name	Type	Description
`type_`	`Literal[FOREVER]`	Type of the trigger. It is a string, which is used to identify the trigger's name. You can change the type by using `Generic` type when inheriting from `BaseTrigger`.

LoopController ¶

Bases: BaseTrigger, ABC, Generic[TriggerTypeT]

Base class for all triggers that have loop control.

Attributes:

Name	Type	Description
`type_`	`TriggerTypeT`	Type of the trigger. It is a string, which is used to identify the trigger's name. You can change the type by using `Generic` type when inheriting from `LoopController`.
`max_loop_count`	`Union[PositiveInt, None]`	The maximum number of times the event should be triggered. If set to 3, then 4th time the event will not be triggered. If set to None, it will keep running forever. This is available for all triggers that inherit from `LoopController`.
`_current_loop_count`	`int`	Current loop count, which is used to keep track of the number of times the event has been triggered. Private attribute, should not be accessed directly. This is available for all triggers that inherit from `LoopController`.

Once ¶

Bases: LoopController[Literal[ONCE]]

A trigger that is triggered only once. It is used to trigger the event only once, and then stop.

Example

from aioclock import AioClock, Once
app = AioClock()

app.task(trigger=Once())
async def task():
    print("Hello World!")

OnStartUp ¶

Bases: LoopController[Literal[ON_START_UP]]

Just like Once, but it triggers the event only once, when the application starts up.

Example

from aioclock import AioClock, OnStartUp
app = AioClock()

app.task(trigger=OnStartUp())
async def task():
    print("Hello World!")

OnShutDown ¶

Bases: LoopController[Literal[ON_SHUT_DOWN]]

Just like Once, but it triggers the event only once, when the application shuts down.

Example

from aioclock import AioClock, OnShutDown
app = AioClock()

app.task(trigger=OnShutDown())
async def task():
    print("Hello World!")

Every ¶

Bases: LoopController[Literal[EVERY]]

A trigger that is triggered every x time units.

Example

from aioclock import AioClock, Every
app = AioClock()

app.task(trigger=Every(seconds=3))
async def task():
    print("Hello World!")

Attributes:

Name	Type	Description
`first_run_strategy`	`Literal['immediate', 'wait']`	Strategy to use for the first run. If `immediate`, then the event will be triggered immediately, and then wait for the time to trigger the event again. If `wait`, then the event will wait for the time to trigger the event for the first time.
`seconds`	`Union[PositiveNumber, None]`	Seconds to wait before triggering the event.
`minutes`	`Union[PositiveNumber, None]`	Minutes to wait before triggering the event.
`hours`	`Union[PositiveNumber, None]`	Hours to wait before triggering the event.
`days`	`Union[PositiveNumber, None]`	Days to wait before triggering the event.
`weeks`	`Union[PositiveNumber, None]`	Weeks to wait before triggering the event.
`max_loop_count`	`Union[PositiveInt, None]`	The maximum number of times the event should be triggered. If set to 3, then 4th time the event will not be triggered. If set to None, it will keep running forever. This is available for all triggers that inherit from `LoopController`.

At ¶

Bases: LoopController[Literal[AT]]

A trigger that is triggered at a specific time.

Example

from aioclock import AioClock, At

app = AioClock()

@app.task(trigger=At(hour=12, minute=30, tz="Asia/Kolkata"))
async def task():
    print("Hello World!")

Attributes:

Name	Type	Description
`second`	`Annotated[int, Interval(ge=0, le=59)]`	Second to trigger the event.
`minute`	`Annotated[int, Interval(ge=0, le=59)]`	Minute to trigger the event.
`hour`	`Annotated[int, Interval(ge=0, le=24)]`	Hour to trigger the event.
`at`	`Literal['every monday', 'every tuesday', 'every wednesday', 'every thursday', 'every friday', 'every saturday', 'every sunday', 'every day']`	Day of week to trigger the event. You would get the in-line typing support when using the trigger.
`tz`	`str`	Timezone to use for the event.
`max_loop_count`	`Union[PositiveInt, None]`	The maximum number of times the event should be triggered. If set to 3, then 4th time the event will not be triggered. If set to None, it will keep running forever. This is available for all triggers that inherit from `LoopController`.

Cron ¶

Bases: LoopController[Literal[CRON]]

A trigger that is triggered at a specific time, using cron job format. If you are not familiar with the cron format, you may read about in this wikipedia article. Or if you need an online tool to generate cron job, you may use crontab.guru.

Example

from aioclock import AioClock, Cron

app = AioClock()

@app.task(trigger=Cron(cron="0 12 * * *", tz="Asia/Kolkata"))
async def task():
    print("Hello World!")

Attributes:

Name	Type	Description
`cron`	`str`	Cron job format to trigger the event.
`tz`	`str`	Timezone to use for the event.
`max_loop_count`	`Union[PositiveInt, None]`	The maximum number of times the event should be triggered. If set to 3, then 4th time the event will not be triggered. If set to None, it will keep running forever. This is available for all triggers that inherit from `LoopController`.

OrTrigger ¶

Bases: LoopController[Literal[OR]]

A trigger that triggers the event if any of the inner triggers are met.

Example

from aioclock import AioClock, OrTrigger, Every, At

app = AioClock()

@app.task(trigger=OrTrigger(triggers=[Every(seconds=3), At(hour=12, minute=30, tz="Asia/Kolkata")]))
async def task():
    print("Hello World!")

Not that any trigger used with OrTrigger, is fully respected, hence if you have two trigger with max_loop_count=1, then each trigger will be triggered only once, and then stop, which result in the OrTrigger run only twice. Check example to understand this intended behaviour.

Example

from aioclock import AioClock, OrTrigger, Every, At

app = AioClock()

@app.task(trigger=OrTrigger( # this get triggered 20 times because :...
    triggers=[
        Every(seconds=3, max_loop_count=10), # will trigger the event 10 times
        At(hour=12, minute=30, tz="Asia/Kolkata", max_loop_count=10) # will trigger the event 10 times
    ]
))
async def task():
    print("Hello World!")

Attributes:

Name	Type	Description
`triggers`	`list[TriggerT]`	List of triggers to use.
`max_loop_count`	`Union[PositiveInt, None]`	The maximum number of times the event should be triggered. If set to 3, then 4th time the event will not be triggered. If set to None, it will keep running forever. This is available for all triggers that inherit from `LoopController`.

Triggers¶

BaseTrigger ¶

trigger_next abstractmethod async ¶

should_trigger ¶

get_waiting_time_till_next_trigger abstractmethod async ¶

Forever ¶

LoopController ¶

Once ¶

OnStartUp ¶

OnShutDown ¶

Every ¶

At ¶

Cron ¶

OrTrigger ¶

trigger_next `abstractmethod` `async` ¶

get_waiting_time_till_next_trigger `abstractmethod` `async` ¶