An asynchronous timer with a human-friendly API and rich functionality.

• State management with run(), pause(), and reset() methods.
• On-the-fly adjustment of the duration with set(), prolong(), and shorten() methods.
• Multi-interval configuration when the timer runs multiple times with a predefined pattern of durations.
• Looping capabilities for continuously running timers.
• Rich callback system enabling hooking into the timer lifecycle events.
• Concurrency-safe architecture designed to prevent race conditions and deadlocks.
• Support for a wide range of Python versions from 3.9 onward.
• Zero third-party dependencies.

• Basic usage
  • One-off timer
  • Multi-interval timer
  • Other usage examples
• States and transitions
• Interval Generator Factories
• Event system
  • Interval complete event
  • Timer complete event
  • Error event
• Advanced usage
  • Configuring precision
  • Custom intervals
• Contributing

from asyncio import run, sleep

from aiotimer import Timer
from aiotimer.interval import once


async def main() -> None:
    """
    Will run the timer for 3 seconds.
    Then will print a message.
    """

    timer = Timer(once(3), lambda: print('3 seconds passed'))
    await timer.run()

    # Wait for the timer to complete.
    await sleep(3 + 1)

if __name__ == '__main__':
    run(main())

from asyncio import run, sleep

from aiotimer import Timer
from aiotimer.interval import thrice


async def main() -> None:
    """
    Will run the timer three times for 1 second each.
    And will print intermediate messages every second.
    Then will print the final message after a total of 3 seconds.
    """

    timer = Timer(
        thrice(1),
        on_timer_complete=lambda: print('3 seconds passed'),
        on_interval_complete=lambda: print('1 more second passed'),
    )
    await timer.run()

    # Wait for the timer to complete.
    await sleep(3 + 1)

if __name__ == '__main__':
    run(main())

More usage examples are available here.

The timer class implements the State Pattern. Methods that modify the timer state may only be called when the timer is in a supported state.

Any transition not listed in the diagram will raise an InvalidStateError. For example, you cannot reset() a timer while it is in the InitialState, and you cannot run() a timer that is in the CompleteState.

This design is used as a defensive programming technique that helps catch any logic errors in the code early and simplifies the debugging process.

Interval Generator Factories (IGFs) are responsible for the making of Interval Generators. Which in turn are responsible for the generation of interval durations for timers.

There are many built-in IGFs that should cover the majority of common use cases.

from aiotimer.interval import *

# Generates 1 interval of 5 seconds.
once(5)


# Generates 2 intervals of 5 seconds each.
twice(5)


# Generates 3 intervals of 5 seconds each.
thrice(5)


# Generates 1 interval between 5 and 10 seconds.
randomly(5, 10)


# Generates 3 intervals of 1, 2, and 3 seconds.
sequentially(1, 2, 3)


# Generates 5 intervals of: 1, 2, 4, 8, and 16 seconds (powers of 2).
exponentially(2, interval_count=5)


# Generates 5 intervals of: 1, 2, 4, 8, and 16 seconds (powers of 2).
exponentially(2, maximum_duration=16)


# Generates 30 intervals of 1, 2, 3, 1, 2, 3, ... seconds.
# Any IGF may be passed as the first argument.
repeatedly(sequentially(1, 2, 3), 10)


# Generates an infinite number of intervals of 1, 2, 3, 1, 2, 3, ... seconds.
# Any IGF may be passed as the first argument.
forever(sequentially(1, 2, 3))


# Generates 3 intervals of 5±0.5 seconds (10% relative jitter).
# Any IGF may be passed as the first argument.
jittery(thrice(5), relative=0.1)


# Generates 3 intervals of 5±0.5 seconds (0.5 second absolute jitter).
# Any IGF may be passed as the first argument.
jittery(thrice(5), absolute=0.5)


# Generates 4 intervals of 0, 5, 5, and 5 seconds.
# Any IGF may be passed as the first argument.
immediately_then(thrice(5))


# Generates no intervals.
# Used in the test suite for testing edge cases. 
never()

If you believe some type of Interval Generator Factory is missing, feel free to submit an issue or a pull request.

All event handlers must comply with the following API contract. Non-compliant event handlers result in undefined behavior.

• Event handler may be either:
  • Synchronous callable.
  • Asynchronous callable.
• Event handler must have either:
  • Zero parameters.
  • Exactly one positional parameter accepting the corresponding event object type.
• Event handler should not return any values because they will be ignored and discarded by the timer.
• An event handler's signature must not be modified at runtime after registration with the timer object.

Any public method of a timer object may be safely called from any event handler. The internal timer architecture prevents any race conditions and deadlocks from occurring.

This event is fired each time any interval of a timer is complete. An on_interval_complete handler may optionally accept an IntervalCompleteEvent object.

This event is fired each time the last interval of a timer is complete. An on_timer_complete handler may optionally accept a TimerCompleteEvent object.

This event is fired each time any exception is propagated from any of the event handlers described above. Additionally, it is fired when an exception occurs inside a system coroutine of a timer. An on_error handler may optionally accept an ErrorEvent object.

The timer class has a configurable precision: float parameter. It represents the amount of seconds a timer would idle between its system ticks.

For adequate accuracy, it is recommended to have the precision value configured significantly (at least several times) smaller than the shortest interval the timer would have.

At the same time, having the precision configured to an extremely low value (e.g. 0.001) may yield a high CPU load.

The first argument to the timer constructor is an Interval Generator Factory. In other words, it is a callable that returns a generator that yields interval durations.

This design decision is required to support multiple functionalities.

A regular list of interval durations could not be used because this would not allow having an infinite number of intervals.

A regular generator object could not be used because it could not be reset to its initial state, which is required to support the timer.reset() functionality.

from asyncio import run, sleep

from aiotimer import Timer


async def main() -> None:
    # The simplest form of the Interval Generator Factory.
    intervals = lambda: (duration for duration in [1, 2, 3])

    timer = Timer(intervals, lambda: print('6 seconds passed'))
    await timer.run()

    # Add an extra second of sleep to avoid a race condition.
    await sleep(6 + 1)

if __name__ == '__main__':
    run(main())

# Create and activate a virtual environment.
python -m venv .
source bin/activate

# Install the library and its dependencies.
pip install --upgrade pip
pip install --editable ".[development]"

# Run the test suite.
BEARTYPE=Yes python -m test --skip-slow=No

Additionally, a convenient Test run configuration is provided for PyCharm users.