Joke

elite.py

import inspect
import asyncio
import functools
from typing import Callable, TypeVar, Union, NewType, Coroutine, Awaitable, Any


__all__ = ["timer", "WaitTime", "Callback", "ReturnType"]

T = TypeVar("T")

WaitTime = NewType("WaitTime", Union[int, float])
SyncCallback = NewType("SyncCallback", Union[Callable[..., T], functools.partial])
AsyncCallback = NewType("Callback", Union[Awaitable, Coroutine])
Callback = NewType("Callback", Union[SyncCallback, AsyncCallback])
ReturnType = NewType("ReturnType", Union[T, Any])


async def waiter(callback: SyncCallback, wait_time: WaitTime) -> ReturnType:
    loop = asyncio.get_running_loop()
    await asyncio.sleep(wait_time)
    return await loop.run_in_executor(None, callback)


async def async_waiter(callback: AsyncCallback, wait_time: WaitTime) -> ReturnType:
    await asyncio.sleep(wait_time)
    return await callback


async def timer(callback: Callback, wait_time: WaitTime) -> ReturnType:
    """Schedule callback to be called
    after the given wait_time number
    of seconds (can be either an int or a float).

    Arguments:
        callback {Callback} -- Your function.
        wait_time {WaitTime} -- Time in seconds.

    Returns:
        ReturnType -- Result from your function.

    Example:

        import asyncio

        from elite import timer

        asyncio.run(timer(functools.partial(func, *args), 10))
        # schedule async function
        asyncio.run(timer(async_func(*args), 10))
        # or
        asyncio.run(timer(another_async_func, 10))
    """
    if inspect.iscoroutinefunction(callback):
        cb = callback()
    elif inspect.isawaitable(callback):
        cb = callback
    elif inspect.isfunction(callback) or isinstance(callback, functools.partial):
        return await waiter(callback, wait_time)
    else:
        raise ValueError(
            "callback can only be of type Awaitable, Coroutine, Callable or functools.partial"
        )
    return await async_waiter(cb, wait_time)