Try to write own async-sync ttl cache decorator (danger: shitcode)

cache_result_by_args.py

import asyncio
import functools
import time


def cached_by_args(sync_or_async_func):
    expiration_offset = 1.5
    cache = {}

    def is_cache_actual(args, kwargs, now) -> bool:
        expire_time, result = cache.get((args, tuple(kwargs.items())), (float('-inf'), None))
        return expire_time >= now

    def get_cached(args, kwargs):
        expire_time, result = cache[(args, tuple(kwargs.items()))]
        return result

    def set_cached(args, kwargs, now, result) -> None:
        cache[(args, tuple(kwargs.items()))] = (now + expiration_offset, result)

    @functools.wraps(sync_or_async_func)
    def sync_wrapper(*args, **kwargs):
        now = time.time()
        if is_cache_actual(args, kwargs, now):
            result = get_cached(args, kwargs)
        else:
            result = sync_or_async_func(*args, **kwargs)
            set_cached(args, kwargs, now, result)
        return result

    @functools.wraps(sync_or_async_func)
    async def async_wrapper(*args, **kwargs):
        now = time.time()
        if is_cache_actual(args, kwargs, now):
            result = get_cached(args, kwargs)
        else:
            result = await sync_or_async_func(*args, **kwargs)
            set_cached(args, kwargs, now, result)
        return result

    if not asyncio.iscoroutinefunction(sync_or_async_func):
        return sync_wrapper
    else:
        return async_wrapper


def shitcode_cached_by_args(sync_or_async_func):
    """
    Async-sync decorator using single wrapper
    (without separate synchronous and asynchronous wrappers).

    WARN: (!) shitcode (!)
    Don't use this in real work!
    """
    expiration_offset = 1.5
    cache = {}

    def is_cache_actual(args, kwargs, now) -> bool:
        expire_time, result = cache.get((args, tuple(kwargs.items())), (float('-inf'), None))
        return expire_time >= now

    def get_cached(args, kwargs):
        expire_time, result = cache[(args, tuple(kwargs.items()))]
        return result

    def set_cached(args, kwargs, now, result) -> None:
        cache[(args, tuple(kwargs.items()))] = (now + expiration_offset, result)

    @functools.wraps(sync_or_async_func)
    def universal_wrapper(*args, **kwargs):
        def result_or_coro():
            nonlocal result
            if not asyncio.iscoroutinefunction(sync_or_async_func):
                return result

            async def f():
                return result
            return f()

        now = time.time()
        if is_cache_actual(args, kwargs, now):
            result = get_cached(args, kwargs)
            return result_or_coro()

        if not asyncio.iscoroutinefunction(sync_or_async_func):
            result = sync_or_async_func(*args, **kwargs)
            set_cached(args, kwargs, now, result)
            return result_or_coro()

        else:
            async def tmp():
                nonlocal result
                result = await sync_or_async_func(*args, **kwargs)
                set_cached(args, kwargs, now, result)
                return result
            return tmp()

    return universal_wrapper


@cached_by_args
async def async_f(*args, **kwargs):
    print(f'CORO RUN {args=} {kwargs=}')
    return 1


@cached_by_args
def sync_f(*args, **kwargs):
    print(f'FUNC RUN {args=} {kwargs=}')
    return 1


async def main():
    sync_f(1)
    sync_f(1)
    sync_f(2)
    await async_f(1)
    await async_f(1)
    await async_f(2)
    print('-' * 5)
    await asyncio.sleep(1.6)
    sync_f(3)
    await async_f(3)
    await asyncio.sleep(1.6)
    sync_f(3)
    await async_f(3)


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