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)
-
Expand source code
def at_interval(interval, func=None, args=(), start=True, loop=None): return Timer(interval, func=func, args=args, start=start, loop=loop)
def datetime_now() ‑> datetime.datetime
-
Expand source code
def datetime_now() -> datetime: return datetime.now(timezone.utc)
Classes
class LazyIntervalTimer (interval_func, func=None, args=(), start=False, loop=None)
-
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()
A timer that calls a function to get the next interval
Ancestors
Inherited members
class Timer (interval, func=None, args=(), start=False, loop=None)
-
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)}>"
Schedules a function to be called asynchronously on a fixed interval
Subclasses
Methods
def call_func(self, *args, **kwargs)
-
Expand source code
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)
Called. Take care of exceptions using gather
def call_next(self)
-
Expand source code
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()
Set next hop in the loop. Call task
def get_delay(self)
-
Expand source code
def get_delay(self): """Return next interval to wait between calls""" return self.interval
Return next interval to wait between calls
def set_result(self, result)
-
Expand source code
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
Set future's result if needed (can be an exception). Else raise if needed.
def start(self)
-
Expand source code
def start(self): """Start scheduling""" self.stop() self.handle = self.loop.call_later(self.get_delay(), self.call_next)
Start scheduling
def stop(self)
-
Expand source code
def stop(self): """Stop scheduling""" if self.handle is not None: self.handle.cancel() self.handle = self.future = None
Stop scheduling