INNER CODE UNIT · Python

_calc_entry_async

python-cachier/cachier · src/cachier/core.py:139

async def _calc_entry_async(core: _BaseCore, key, func, args, kwds, printer=lambda *_: None) -> Optional[Any]:
    await core.amark_entry_being_calculated(key)
    try:
        func_res = await func(*args, **kwds)
        stored = await core.aset_entry(key, func_res)
        if not stored:
            printer("Result exceeds entry_size_limit; not cached")
        return func_res
    finally:
        await core.amark_entry_not_calculated(key)


def _convert_args_kwargs(func, _is_method: bool, args: tuple, kwds: dict) -> dict:
    """Convert mix of positional and keyword arguments to aggregated kwargs."""
    # unwrap if the function is functools.partial
    if hasattr(func, "func"):
        args = func.args + args
        kwds.update({k: v for k, v in func.keywords.items() if k not in kwds})

View source record →