A decorator for calling the decorated function at a specific interval

call_every.py

# Requires:
#  pip install fastapi
#  pip install psutil
import asyncio
import time
from typing import Callable
import os
import psutil
import logging
import warnings
from fastapi.concurrency import run_in_threadpool

logger = logging.getLogger(__name__)


def call_every(
    seconds: float,
    wait_first: bool = False,
):
    def timed_task_decorator(func: Callable) -> Callable:

        is_coroutine = asyncio.iscoroutinefunction(func)

        async def timer():

            if wait_first:
                await asyncio.sleep(seconds)

            while True:
                t_0 = time.time()

                try:
                    if is_coroutine:
                        await func()
                    else:
                        await run_in_threadpool(func)
                except Exception:  # pylint: disable=broad-except
                    logger.exception(f"Exception in {func}")

                remainder = seconds - (time.time() - t_0)

                if remainder < 0:
                    warnings.warn(
                        f"Function {func} has an execution time the exceeds"
                        f" the requested execution interval of {seconds}s!",
                        UserWarning,
                    )

                await asyncio.sleep(max(remainder, 0))

        # Put `timer` on the event loop on service startup
        asyncio.ensure_future(timer())
        return func

    return timed_task_decorator


if __name__ == "__main__":
  
    # A handle for the python process we are currently running in
    this = psutil.Process(os.getpid())

    @call_every(0.1)
    def do_it():
        # Every 0.1 second -> print the current cpu usage, memory usage and thread count. Overwrites on the same line!
        print(
            f" CPU: {this.cpu_percent():9.5f}%, MEM: {this.memory_percent():9.5f}%, ThreadCount: {this.num_threads()}",
            end="\r",
        )

    # Get the active event loop and run forever
    asyncio.get_event_loop().run_forever()