Module server.timing

Helpers for executing async functions on a timer

Sub-modules

server.timing.timer

This code is a modified version of Gael Pasgrimaud's library aiocron …

Functions

def at_interval(interval, func=None, args=(), start=True, loop=None)

def datetime_now() ‑> datetime.datetime

Classes

class LazyIntervalTimer (interval_func, func=None, args=(), start=False, loop=None)

A timer that calls a function to get the next interval

Expand source code

class LazyIntervalTimer(Timer):
    """A timer that calls a function to get the next interval"""

    def __init__(
        self,
        interval_func,
        func=None,
        args=(),
        start=False,
        loop=None
    ):
        super().__init__(
            interval=None,
            func=func,
            args=args,
            start=start,
            loop=loop
        )
        self.interval_func = interval_func

    def get_delay(self):
        return self.interval_func()

Ancestors

• Timer

Inherited members

• Timer:
  • call_func
  • call_next
  • get_delay
  • set_result
  • start
  • stop

class Timer (interval, func=None, args=(), start=False, loop=None)

Schedules a function to be called asynchronously on a fixed interval

Expand source code

class Timer(object):
    """Schedules a function to be called asynchronously on a fixed interval"""

    def __init__(self, interval, func=None, args=(), start=False, loop=None):
        self.interval = interval
        if func is not None:
            self.func = func if not args else functools.partial(func, *args)
        else:
            self.func = null_callback
        self.cron = wrap_func(self.func)
        self.auto_start = start
        self.handle = self.future = None
        self.loop = loop if loop is not None else asyncio.get_running_loop()
        if start and self.func is not null_callback:
            self.handle = self.loop.call_soon_threadsafe(self.start)

    def start(self):
        """Start scheduling"""
        self.stop()
        self.handle = self.loop.call_later(self.get_delay(), self.call_next)

    def stop(self):
        """Stop scheduling"""
        if self.handle is not None:
            self.handle.cancel()
        self.handle = self.future = None

    def get_delay(self):
        """Return next interval to wait between calls"""
        return self.interval

    def call_next(self):
        """Set next hop in the loop. Call task"""
        if self.handle is not None:
            self.handle.cancel()
        self.handle = self.loop.call_later(self.get_delay(), self.call_next)
        self.call_func()

    def call_func(self, *args, **kwargs):
        """Called. Take care of exceptions using gather"""
        asyncio.gather(
            self.cron(*args, **kwargs),
            return_exceptions=True
        ).add_done_callback(self.set_result)

    def set_result(self, result):
        """Set future's result if needed (can be an exception).
        Else raise if needed."""
        result = result.result()[0]
        if self.future is not None:
            if isinstance(result, Exception):
                self.future.set_exception(result)
            else:
                self.future.set_result(result)
            self.future = None
        elif isinstance(result, Exception):
            raise result

    def __call__(self, func):
        """Used as a decorator"""
        self.func = func
        self.cron = wrap_func(func)
        if self.auto_start:
            self.loop.call_soon_threadsafe(self.start)
        return self

    def __str__(self):
        return f"{self.get_delay()} {self.func}"

    def __repr__(self):
        return f"<Timer {str(self)}>"

Subclasses

• LazyIntervalTimer

Methods

def call_func(self, *args, **kwargs)

Called. Take care of exceptions using gather

def call_next(self)

Set next hop in the loop. Call task

def get_delay(self)

Return next interval to wait between calls

def set_result(self, result)

Set future's result if needed (can be an exception). Else raise if needed.

def start(self)

Start scheduling

def stop(self)

Stop scheduling