Asyncio Timer

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

Features:

• State management with start(), stop(), and reset() methods.
• On-the-fly duration adjustment with prolong(), shorten(), and set() methods.
• Introspection with elapsed, remaining, and state properties.
• Multi-interval timer configurations.
• Continuously running timers.
• Rich callback system enabling hooking into the timer lifecycle events.
• Synchronous and asynchronous callback execution modes.
• Drift-free implementation, accurate regardless of interval length and count.
• Concurrency-safe architecture designed to prevent race conditions and deadlocks.
• Support for a wide range of Python versions from 3.9 onward.
• Zero production dependencies except for typing-extensions by the Python core team.

Table of contents

• Usage examples
  • One-off timer
  • Multi-interval timer
  • Other usage examples
• Public API
  • Controlling the state
  • Duration modification
  • Introspection
• States and transitions
• Configuring durations
  • Single duration
  • Sequence of durations
  • Duration factory
  • Duration multipliers
• Event system
  • Interval complete event
  • Timer complete event
  • Error event
• Advanced usage
  • Sync and Async callbacks
  • Configuring precision
  • Custom duration factories
  • Memory management
  • Runtime type checking
• Contributing

🔝 Usage examples

🔝 One-off timer

A timer may have just one time interval.

from asyncio import run, sleep

from aiotimer import Timer


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

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

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


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

🔝 Multi-interval timer

A timer may have multiple time intervals of arbitrary durations.

from asyncio import run, sleep

from aiotimer import Timer
from aiotimer.duration.factory import thrice
from aiotimer.duration.multiplier import second


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 * second),
    lambda: print('3 seconds passed'),
    lambda: print('1 more second passed'),
  )
  await timer.start()

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


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

🔝 Other usage examples

More usage examples are available here.

🔝 Back to top

🔝 Public API

🔝 Controlling the state

• await timer.start() starts the timer that is in the Initial or in the Stopped state.
• await timer.stop() stops the timer that is in the Running state. The elapsed and remaining times for the current time interval as well as the current interval itself are preserved.
• await timer.reset() resets the timer. The elapsed and remaining times for the current time interval as well as the current interval itself are discarded. The timer is reset to the initial state it had after instantiation.

🔝 Duration modification

• await timer.set(duration) sets the duration of the currently running interval to duration. In case the elapsed time is greater than duration, the interval would complete immediately.
• await timer.prolong(delta) prolongs the duration of the currently running interval by delta.
• await timer.shorten(delta) shortens the duration of the currently running interval by delta. In case the elapsed time is greater than the resulting duration after shortening, the interval would complete immediately.

🔝 Introspection

• timer.elapsed returns the elapsed time for the currently running interval.
• timer.remaining returns the remaining time for the currently running interval.
• timer.state returns the type of the current state of the timer.

🔝 Back to top

🔝 States and transitions

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 start() 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.

🔝 Back to top

🔝 Configuring durations

The first parameter of the timer constructor defines its interval durations. A timer may have one or more time intervals of arbitrary durations, all measured in seconds. The parameter accepts three different forms, from the simplest to the most powerful:

• A single duration — for one-off timers with a single interval.
• A sequence of durations — for timers with a fixed, known set of intervals.
• A duration factory — for complex duration generation logic.

All interval durations must be positive numbers or zeroes, regardless of the form used. A negative duration yields an undefined behavior.

Duration objects must not be modified after they are passed to a timer, regardless of the form used. Modifying them yields an undefined behavior.

🔝 Single duration

This is the simplest duration form for one-off timers. Any integer or floating-point number configures a timer with exactly one interval of the given duration in seconds.

from aiotimer import Timer


# A one-off timer with a single 5-second interval.
Timer(5, lambda: print('Timer complete'))

🔝 Sequence of durations

This is the most convenient duration form when the set of intervals is static and known upfront. Any Sequence of durations (e.g. a list or a tuple) configures a multi-interval timer.

from aiotimer import Timer


# Three intervals of 1, 2, and 3 seconds.
Timer(
  [1, 2, 3],
  lambda: print('Timer complete'),
  lambda: print('Interval complete'),
)


# Same as previous, but using a tuple.
Timer(
  (1, 2, 3),
  lambda: print('Timer complete'),
  lambda: print('Interval complete'),
)

A duration sequence must not be empty. An empty sequence raises an EmptyDurationIterableError.

🔝 Duration factory

This is the most powerful form of durations. A Duration Factory is a callable that returns an Iterable of durations. It powers more complex duration generation logic as well as infinitely-running timers.

The library is shipped with an extensive catalog of built-in duration factories that should cover the majority of common use cases. Also see Custom duration factories for implementing your own factory.

from aiotimer.duration.factory 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 10 intervals of 5 seconds each.
repeatedly(5, 10)


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


# Generates an infinite number of 5-second intervals.
forever(5)


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


# Generates 3 intervals of 5±0.5 seconds (10% relative jitter).
# Any other factory 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 other factory may be passed as the first argument.
jittery(thrice(5), absolute=0.5)


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


# Same as previous but limited by the duration instead.
exponentially(maximum_duration=16)


# Generates faster-growing intervals of 1, 3, and 9 seconds.
exponentially(base=3, interval_count=3)


# Generates scaled-down intervals of 0.1, 0.2, and 0.4 seconds.
exponentially(scale=0.1, interval_count=3)


# Generates a zero-second interval followed by 5 exponentially growing retries.
# Retry delays are powers of 2, each divided by 2 with a ±25% jitter applied.
# Resulting durations are 0, 0.5±25%, 1±25%, 2±25%, 4±25%, and 8±25%.
backoff()


# Generates more retries.
backoff(retries=10)


# Stops retry generation after reaching the maximum duration.
# Jitter is excluded from the calculation.
backoff(maximum_duration=60)


# Applies a faster exponential growth.
backoff(base=3)


# Applies a stronger down-scaling.
backoff(scale=0.1)


# Applies a lighter jitter.
backoff(jitter=0.1)


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

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

All factories that wrap another duration source (once, twice, thrice, repeatedly, sequentially, forever, jittery, immediately_then) accept all three forms of durations. This enables a very powerful mechanism of factory composition. A few examples are listed below.

from aiotimer.duration.factory import *


# Generates 9 intervals with a repeating pattern of 1, 2, and 3 seconds.
thrice([1, 2, 3])


# Generates 10 intervals between 5 and 10 seconds each.
repeatedly(randomly(5, 10), 10)


# Generates a random interval followed by an exponential sequence.
sequentially(
  randomly(0.1, 0.2),
  exponentially(interval_count=10),
)


# Generates 10 exponentially-growing intervals with a 10% jitter applied to them.
jittery(exponentially(interval_count=10), relative=0.1)


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

🔝 Duration multipliers

A timer always expects durations to be passed in seconds (float or int). To express durations in other time units conveniently, the library ships a set of duration multipliers — plain numeric constants that scale duration values into seconds.

from aiotimer.duration.multiplier import *


millisecond, milliseconds  # 0.001
second, seconds            # 1
minute, minutes            # 60
hour, hours                # 3600
day, days                  # 86400
week, weeks                # 604800
month, months              # 2592000 (30 days)
year, years                # 31536000 (365 days)

Each multiplier is available in both a singular and a plural form. The two are interchangeable. You may pick whichever reads more naturally.

from aiotimer import Timer
from aiotimer.duration.factory import thrice
from aiotimer.duration.multiplier import hour, milliseconds, minutes, seconds

# A single 5-minute interval.
Timer(5 * minutes)

# Three intervals of 30 seconds, 5 minutes, and 1 hour.
Timer([30 * seconds, 5 * minutes, 1 * hour])

# Three 100-millisecond intervals.
Timer(thrice(100 * milliseconds))

🔝 Back to top

🔝 Event system

There are several event handlers that may be configured for a timer through the constructor arguments.

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

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

All event objects have a timer property that references the timer object that fired the event.

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.

🔝 Interval complete event

This event is fired each time any interval of a timer is complete. An on_interval_complete handler may optionally accept an IntervalCompleteEvent object. Events of this type have the following properties:

• timer: Timer
• interval_number: int
• interval_duration: float

🔝 Timer complete event

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. Events of this type have the following properties:

• timer: Timer
• interval_count: int

🔝 Error event

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. Events of this type have the following properties:

• timer: Timer
• error: Exception

🔝 Back to top

🔝 Advanced usage

🔝 Sync and Async callbacks

Use the await_callbacks parameter of the Timer constructor to control the way the callbacks are handled.

In the sync mode (await_callbacks == True) the next interval would not start until the on_interval_complete callback finishes execution.

In the async mode (await_callbacks == False) the next interval would start immediately after the previous one completes.

Both modes support def, async def as well as any other types of compatible callables. It's perfectly fine to use def in the async mode and async def in sync mode.

🔝 Configuring precision

The timer class has a configurable precision: float parameter. It represents the number 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.

🔝 Custom duration factories

A Duration Factory is one of the three forms the timer constructor accepts for its first argument (see Configuring durations). It is a callable that returns an Iterable of durations.

In case the built-in factories do not cover your usage scenarios, you can construct your custom one. The simplest custom duration factory is a lambda returning a list of durations.

from asyncio import run, sleep

from aiotimer import Timer


async def main() -> None:
  duration_factory = lambda: [1, 2, 3]

  timer = Timer(
    duration_factory,
    lambda: print('6 seconds passed'),
  )

  await timer.start()

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


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

🔝 Memory management

A timer in the Running state will never be garbage-collected, nor will event handlers registered with it. They are referenced by the event loop and live at least until the timer is stopped.

Inherently infinitely-running timers must be stopped manually as soon as they are no longer needed. Failing to do so effectively results in a memory leak.

🔝 Runtime type checking

The library supports optional runtime type checking of its whole codebase powered by beartype.

To enable it, install beartype and set the BEARTYPE environment variable to any truthy value (e.g. Yes, True, 1) before importing the library.

pip install beartype
BEARTYPE=Yes python main.py

The variable name intentionally does not have a library prefix, so that a single switch can enable runtime type checking across every library and application following the same convention.

In case the variable is set but beartype is not installed, the library emits a warning and runs normally, just with type checking disabled.

🔝 Back to top

🔝 Contributing

🔝 Configuring the development environment

# 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, convenient Quick QA and Full QA run configurations are provided for PyCharm users.

🔝 Back to top