core.py

"""Persistent, stale-free memoization decorators for Python."""

# This file is part of Cachier.
# https://github.com/python-cachier/cachier

# Licensed under the MIT license:
# http://www.opensource.org/licenses/MIT-license
# Copyright (c) 2016, Shay Palachy <shaypal5@gmail.com>

import asyncio
import inspect
import os
import threading
import warnings
from collections import OrderedDict
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta
from functools import wraps
from typing import Any, Callable, Optional, ParamSpec, Protocol, TypeVar, Union
from warnings import warn

from ._types import RedisClient, S3Client
from .config import Backend, HashFunc, Mongetter, _update_with_defaults
from .cores.base import RecalculationNeeded, _BaseCore
from .cores.memory import _MemoryCore
from .cores.mongo import _MongoCore
from .cores.pickle import _PickleCore
from .cores.redis import _RedisCore
from .cores.s3 import _S3Core
from .cores.sql import _SQLCore
from .metrics import CacheMetrics, MetricsContext
from .util import parse_bytes

_P = ParamSpec("_P")
_R = TypeVar("_R")
_R_co = TypeVar("_R_co", covariant=True)


class _CachierWrappedFunc(Protocol[_P, _R_co]):
    """Callable returned by ``@cachier`` with the decorated function's signature.

    Preserves the original function's parameter and return types via ``ParamSpec`` while also exposing the cache-
    management attributes attached by the decorator. Per-call cachier options such as ``max_age`` and
    ``cachier__skip_cache`` are accepted at runtime but are not surfaced in the ``__call__`` signature here; PEP 612
    does not permit mixing ParamSpec kwargs with additional keyword-only parameters.

    """

    def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _R_co: ...  # pragma: no cover

    clear_cache: Callable[..., Any]
    clear_being_calculated: Callable[[], Any]
    aclear_cache: Callable[..., Any]
    aclear_being_calculated: Callable[[], Any]
    cache_dpath: Callable[[], Optional[str]]
    precache_value: Callable[..., Any]
    metrics: Optional[CacheMetrics]


MAX_WORKERS_ENVAR_NAME = "CACHIER_MAX_WORKERS"
DEFAULT_MAX_WORKERS = 8
ZERO_TIMEDELTA = timedelta(seconds=0)


class _ImmediateAwaitable:
    """Lightweight awaitable that yields an immediate value."""

    def __init__(self, value: Any = None) -> None:
        self._value = value

    def __await__(self):
        if False:
            yield None
        return self._value


async def _background_recalc_async(
    core: _BaseCore,
    key: Any,
    func: Callable[..., Any],
    args: Any,
    kwds: Any,
) -> None:
    """Run async recomputation in background and clear processing flag.

    This helper ensures that the cache entry's "being calculated" state is cleared only after the background
    recomputation and cache update (performed by ``_function_thread_async``) have completed.

    """
    try:
        await _function_thread_async(core, key, func, args, kwds)
    finally:
        await core.amark_entry_not_calculated(key)


def _max_workers():
    return int(os.environ.get(MAX_WORKERS_ENVAR_NAME, DEFAULT_MAX_WORKERS))


def _set_max_workers(max_workers):
    os.environ[MAX_WORKERS_ENVAR_NAME] = str(max_workers)
    _get_executor(True)


def _get_executor(reset=False):
    if reset or not hasattr(_get_executor, "executor"):
        _get_executor.executor = ThreadPoolExecutor(_max_workers())
    return _get_executor.executor


def _function_thread(core: _BaseCore, key, func, args, kwds):
    try:
        func_res = func(*args, **kwds)
        core.set_entry(key, func_res)
    except BaseException as exc:
        print(f"Function call failed with the following exception:\n{exc}")


async def _function_thread_async(core: _BaseCore, key, func, args, kwds):
    try:
        func_res = await func(*args, **kwds)
        await core.aset_entry(key, func_res)
    except BaseException as exc:
        print(f"Function call failed with the following exception:\n{exc}")


def _calc_entry(core: _BaseCore, key, func, args, kwds, printer=lambda *_: None) -> Optional[Any]:
    core.mark_entry_being_calculated(key)
    try:
        func_res = func(*args, **kwds)
        stored = core.set_entry(key, func_res)
        if not stored:
            printer("Result exceeds entry_size_limit; not cached")
        return func_res
    finally:
        core.mark_entry_not_calculated(key)


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})
        func = func.func

    sig = inspect.signature(func)
    func_params = list(sig.parameters)

    # Separate regular parameters from VAR_POSITIONAL
    regular_params = []
    var_positional_name = None

    for param_name in func_params:
        param = sig.parameters[param_name]
        if param.kind == inspect.Parameter.VAR_POSITIONAL:
            var_positional_name = param_name
        elif param.kind in (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD):
            regular_params.append(param_name)

    # Map positional arguments to regular parameters
    if _is_method:
        # Skip 'self' for methods
        args_to_map = args[1:]
        params_to_use = regular_params[1:]
    else:
        args_to_map = args
        params_to_use = regular_params

    # Map as many args as possible to regular parameters
    num_regular = len(params_to_use)
    args_as_kw = {params_to_use[index]: arg for index, arg in enumerate(args_to_map[:num_regular])}

    # Handle variadic positional arguments
    # Store them with indexed keys like __varargs_0__, __varargs_1__, etc.
    if var_positional_name and len(args_to_map) > num_regular:
        var_args = args_to_map[num_regular:]
        for i, arg in enumerate(var_args):
            args_as_kw[f"__varargs_{i}__"] = arg

    # Init with default values
    kwargs = {k: v.default for k, v in sig.parameters.items() if v.default is not inspect.Parameter.empty}

    # Merge args expanded as kwargs and the original kwds
    kwargs.update(args_as_kw)

    # Handle keyword arguments (including variadic keyword arguments)
    kwargs.update(kwds)

    return OrderedDict(sorted(kwargs.items()))


def _pop_kwds_with_deprecation(kwds, name: str, default_value: bool):
    if name in kwds:
        warnings.warn(
            f"`{name}` is deprecated and will be removed in a future release, use `cachier__` alternative instead.",
            DeprecationWarning,
            stacklevel=2,
        )
    return kwds.pop(name, default_value)


def _is_async_redis_client(client: Any) -> bool:
    if client is None:
        return False
    method_names = ("hgetall", "hset", "keys", "delete", "hget")
    return all(inspect.iscoroutinefunction(getattr(client, name, None)) for name in method_names)


def _convert_public_cache_args(func, _is_method: bool, args: tuple, kwds: dict) -> dict:
    """Convert cache-management arguments to canonical cache-key kwargs."""
    if _is_method:
        args = (None, *args)
    return _convert_args_kwargs(func, _is_method=_is_method, args=args, kwds=kwds)


def cachier(
    hash_func: Optional[HashFunc] = None,
    hash_params: Optional[HashFunc] = None,
    backend: Optional[Backend] = None,
    mongetter: Optional[Mongetter] = None,
    sql_engine: Optional[Union[str, Any, Callable[[], Any]]] = None,
    redis_client: Optional["RedisClient"] = None,
    s3_bucket: Optional[str] = None,
    s3_prefix: str = "cachier",
    s3_client: Optional["S3Client"] = None,
    s3_client_factory: Optional[Callable[[], Any]] = None,
    s3_region: Optional[str] = None,
    s3_endpoint_url: Optional[str] = None,
    s3_config: Optional[Any] = None,
    stale_after: Optional[timedelta] = None,
    next_time: Optional[bool] = None,
    cache_dir: Optional[Union[str, os.PathLike]] = None,
    pickle_reload: Optional[bool] = None,
    separate_files: Optional[bool] = None,
    wait_for_calc_timeout: Optional[int] = None,
    allow_none: Optional[bool] = None,
    cleanup_stale: Optional[bool] = None,
    cleanup_interval: Optional[timedelta] = None,
    entry_size_limit: Optional[Union[int, str]] = None,
    allow_non_static_methods: Optional[bool] = None,
    enable_metrics: bool = False,
    metrics_sampling_rate: float = 1.0,
) -> Callable[[Callable[_P, _R]], _CachierWrappedFunc[_P, _R]]:
    """Wrap as a persistent, stale-free memoization decorator.

    The positional and keyword arguments to the wrapped function must be
    hashable (i.e. Python's immutable built-in objects, not mutable
    containers). Also, notice that since objects which are instances of
    user-defined classes are hashable but all compare unequal (their hash
    value is their id), equal objects across different sessions will not yield
    identical keys.

    Parameters
    ----------
    hash_func : callable, optional
        A callable that gets the args and kwargs from the decorated function
        and returns a hash key for them. This parameter can be used to enable
        the use of cachier with functions that get arguments that are not
        automatically hashable by Python.
    hash_params : callable, optional
        Deprecated, use :func:`~cachier.core.cachier.hash_func` instead.
    backend : str, optional
        The name of the backend to use. Valid options currently include
        'pickle', 'mongo', 'memory', 'sql', 'redis', and 's3'. If not
        provided, defaults to 'pickle', unless a core-associated parameter
        is provided.
    mongetter : callable, optional
        A callable that takes no arguments and returns a pymongo.Collection
        object with writing permissions. If provided, the backend is set to
        'mongo'.
    sql_engine : str, Engine, or callable, optional
        SQLAlchemy connection string, Engine, or callable returning an Engine.
        Used for the SQL backend.
    redis_client : redis.Redis or callable, optional
        Redis client instance or callable returning a Redis client.
        Used for the Redis backend.
    s3_bucket : str, optional
        The S3 bucket name for cache storage. Required when using the S3 backend.
    s3_prefix : str, optional
        Key prefix applied to all S3 cache objects. Defaults to ``"cachier"``.
    s3_client : boto3 S3 client, optional
        A pre-configured boto3 S3 client instance.
    s3_client_factory : callable, optional
        A callable that returns a boto3 S3 client, allowing lazy initialization.
    s3_region : str, optional
        AWS region name used when auto-creating the boto3 S3 client.
    s3_endpoint_url : str, optional
        Custom endpoint URL for S3-compatible services such as MinIO or localstack.
    s3_config : botocore.config.Config, optional
        Optional botocore Config object passed when auto-creating the client.
    stale_after : datetime.timedelta, optional
        The time delta after which a cached result is considered stale. Calls
        made after the result goes stale will trigger a recalculation of the
        result, but whether a stale or fresh result will be returned is
        determined by the optional next_time argument.
    next_time : bool, optional
        If set to True, a stale result will be returned when finding one, not
        waiting for the calculation of the fresh result to return. Defaults to
        False.
    cache_dir : str, optional
        A fully qualified path to a file directory to be used for cache files.
        The running process must have running permissions to this folder. If
        not provided, a default directory at `~/.cachier/` is used.
    pickle_reload : bool, optional
        If set to True, in-memory cache will be reloaded on each cache read,
        enabling different threads to share cache. Should be set to False for
        faster reads in single-thread programs. Defaults to True.
    separate_files: bool, default False, for Pickle cores only
        Instead of a single cache file per-function, each function's cache is
        split between several files, one for each argument set. This can help
        if your per-function cache files become too large.
    wait_for_calc_timeout: int, optional
        The maximum time to wait for an ongoing calculation. When a
        process started to calculate the value setting being_calculated to
        True, any process trying to read the same entry will wait a maximum of
        seconds specified in this parameter. 0 means wait forever.
        Once the timeout expires the calculation will be triggered.
    allow_none: bool, optional
        Allows storing None values in the cache. If False, functions returning
        None will not be cached and are recalculated every call.
    cleanup_stale: bool, optional
        If True, stale cache entries are periodically deleted in a background
        thread. Defaults to False.
    cleanup_interval: datetime.timedelta, optional
        Minimum time between automatic cleanup runs. Defaults to one day.
    entry_size_limit: int or str, optional
        Maximum serialized size of a cached value. Values exceeding the limit
        are returned but not cached. Human readable strings like ``"10MB"`` are
        allowed.
    allow_non_static_methods : bool, optional
        If True, allows ``@cachier`` to decorate instance methods (functions
        whose first parameter is named ``self``). By default, decorating an
        instance method raises ``TypeError`` because the ``self`` argument is
        ignored for cache-key computation, meaning all instances share the
        same cache -- which is rarely the intended behaviour. Set this to
        ``True`` only when cross-instance cache sharing is intentional.
    enable_metrics: bool, optional
        Enable metrics collection for this cached function. When enabled,
        cache hits, misses, latencies, and other performance metrics are tracked. Defaults to False.
    metrics_sampling_rate: float, optional
        Sampling rate for metrics collection (0.0 to 1.0). Lower values
        reduce overhead at the cost of accuracy. Only used when enable_metrics is True. Defaults to 1.0 (100% sampling).

    """
    # Check for deprecated parameters
    if hash_params is not None:
        message = "hash_params will be removed in a future release, please use hash_func instead"
        warn(message, DeprecationWarning, stacklevel=2)
        hash_func = hash_params
    # Update parameters with defaults if input is None
    backend = _update_with_defaults(backend, "backend")
    mongetter = _update_with_defaults(mongetter, "mongetter")
    size_limit_bytes = parse_bytes(_update_with_defaults(entry_size_limit, "entry_size_limit"))

    # Create metrics object if enabled
    cache_metrics = None
    if enable_metrics:
        cache_metrics = CacheMetrics(sampling_rate=metrics_sampling_rate)

    # Override the backend parameter if a mongetter is provided.
    if callable(mongetter):
        backend = "mongo"
    core: _BaseCore
    if backend == "pickle":
        core = _PickleCore(
            hash_func=hash_func,
            pickle_reload=pickle_reload,
            cache_dir=cache_dir,
            separate_files=separate_files,
            wait_for_calc_timeout=wait_for_calc_timeout,
            entry_size_limit=size_limit_bytes,
            metrics=cache_metrics,
        )
    elif backend == "mongo":
        core = _MongoCore(
            hash_func=hash_func,
            mongetter=mongetter,
            wait_for_calc_timeout=wait_for_calc_timeout,
            entry_size_limit=size_limit_bytes,
            metrics=cache_metrics,
        )
    elif backend == "memory":
        core = _MemoryCore(
            hash_func=hash_func,
            wait_for_calc_timeout=wait_for_calc_timeout,
            entry_size_limit=size_limit_bytes,
            metrics=cache_metrics,
        )
    elif backend == "sql":
        core = _SQLCore(
            hash_func=hash_func,
            sql_engine=sql_engine,
            wait_for_calc_timeout=wait_for_calc_timeout,
            entry_size_limit=size_limit_bytes,
            metrics=cache_metrics,
        )
    elif backend == "redis":
        core = _RedisCore(
            hash_func=hash_func,
            redis_client=redis_client,
            wait_for_calc_timeout=wait_for_calc_timeout,
            entry_size_limit=size_limit_bytes,
            metrics=cache_metrics,
        )
    elif backend == "s3":
        core = _S3Core(
            hash_func=hash_func,
            s3_bucket=s3_bucket,
            wait_for_calc_timeout=wait_for_calc_timeout,
            s3_prefix=s3_prefix,
            s3_client=s3_client,
            s3_client_factory=s3_client_factory,
            s3_region=s3_region,
            s3_endpoint_url=s3_endpoint_url,
            s3_config=s3_config,
            entry_size_limit=size_limit_bytes,
            metrics=cache_metrics,
        )
    else:
        raise ValueError("specified an invalid core: %s" % backend)

    def _cachier_decorator(func: Callable[_P, _R]) -> _CachierWrappedFunc[_P, _R]:
        core.set_func(func)

        # Guard: raise TypeError when decorating an instance method unless
        # explicitly opted in.  The 'self' parameter is ignored for cache-key
        # computation, so all instances share the same cache.
        if core.func_is_method:
            _allow_methods = _update_with_defaults(allow_non_static_methods, "allow_non_static_methods")
            if not _allow_methods:
                raise TypeError(
                    f"@cachier cannot decorate instance method "
                    f"'{func.__qualname__}' because the 'self' parameter is "
                    "excluded from cache-key computation and all instances "
                    "would share a single cache. Pass allow_non_static_methods=True "
                    "to the decorator or call "
                    "set_global_params(allow_non_static_methods=True) if "
                    "cross-instance cache sharing is intentional."
                )

        is_coroutine = inspect.iscoroutinefunction(func)

        if backend == "mongo":
            if is_coroutine and not inspect.iscoroutinefunction(mongetter):
                msg = "Async cached functions with Mongo backend require an async mongetter."
                raise TypeError(msg)
            if (not is_coroutine) and inspect.iscoroutinefunction(mongetter):
                msg = "Async mongetter requires an async cached function."
                raise TypeError(msg)

        if backend == "redis":
            if is_coroutine:
                if callable(redis_client):
                    if not inspect.iscoroutinefunction(redis_client):
                        msg = "Async cached functions with Redis backend require an async redis_client callable."
                        raise TypeError(msg)
                elif not _is_async_redis_client(redis_client):
                    msg = "Async cached functions with Redis backend require an async Redis client."
                    raise TypeError(msg)
            else:
                if callable(redis_client) and inspect.iscoroutinefunction(redis_client):
                    msg = "Async redis_client callable requires an async cached function."
                    raise TypeError(msg)
                if _is_async_redis_client(redis_client):
                    msg = "Async Redis client requires an async cached function."
                    raise TypeError(msg)

        if backend == "sql":
            sql_core = core
            assert isinstance(sql_core, _SQLCore)  # noqa: S101
            if is_coroutine and not sql_core.has_async_engine():
                msg = "Async cached functions with SQL backend require an AsyncEngine sql_engine."
                raise TypeError(msg)
            if (not is_coroutine) and sql_core.has_async_engine():
                msg = "Async SQL engines require an async cached function."
                raise TypeError(msg)

        last_cleanup = datetime.min
        cleanup_lock = threading.Lock()

        # ---
        # MAINTAINER NOTE: max_age parameter
        #
        # The _call function below supports a per-call 'max_age' parameter,
        # allowing users to specify a maximum allowed age for a cached value.
        # If the cached value is older than 'max_age',
        # a recalculation is triggered. This is in addition to the
        # per-decorator 'stale_after' parameter.
        #
        # The effective staleness threshold is the minimum of 'stale_after'
        # and 'max_age' (if provided).
        # This ensures that the strictest max age requirement is enforced.
        #
        # The main function wrapper is a standard function that passes
        # *args and **kwargs to _call. By default, max_age is None,
        # so only 'stale_after' is considered unless overridden.
        #
        # The user-facing API exposes:
        #   - Per-call: myfunc(..., max_age=timedelta(...))
        #
        # This design allows both one-off (per-call) and default
        # (per-decorator) max age constraints.
        # ---

        def _call(*args, max_age: Optional[timedelta] = None, **kwds):
            nonlocal allow_none, last_cleanup
            _allow_none = _update_with_defaults(allow_none, "allow_none", kwds)
            # print('Inside general wrapper for {}.'.format(func.__name__))
            ignore_cache = _pop_kwds_with_deprecation(kwds, "ignore_cache", False)
            overwrite_cache = _pop_kwds_with_deprecation(kwds, "overwrite_cache", False)
            verbose = _pop_kwds_with_deprecation(kwds, "verbose_cache", False)
            ignore_cache = kwds.pop("cachier__skip_cache", ignore_cache)
            overwrite_cache = kwds.pop("cachier__overwrite_cache", overwrite_cache)
            verbose = kwds.pop("cachier__verbose", verbose)
            _stale_after = _update_with_defaults(stale_after, "stale_after", kwds)
            _next_time = _update_with_defaults(next_time, "next_time", kwds)
            _cleanup_flag = _update_with_defaults(cleanup_stale, "cleanup_stale", kwds)
            _cleanup_interval_val = _update_with_defaults(cleanup_interval, "cleanup_interval", kwds)
            # merge args expanded as kwargs and the original kwds
            kwargs = _convert_args_kwargs(func, _is_method=core.func_is_method, args=args, kwds=kwds)

            if _cleanup_flag:
                now = datetime.now()
                with cleanup_lock:
                    if now - last_cleanup >= _cleanup_interval_val:
                        last_cleanup = now
                        _get_executor().submit(core.delete_stale_entries, _stale_after)

            _print = print if verbose else lambda x: None

            # Check current global caching state dynamically
            from .config import _global_params

            if ignore_cache or not _global_params.caching_enabled:
                return func(args[0], **kwargs) if core.func_is_method else func(**kwargs)  # type: ignore[call-arg]

            with MetricsContext(cache_metrics) as _mctx:
                key, entry = core.get_entry((), kwargs)
                if overwrite_cache:
                    _mctx.record_miss()
                    _mctx.record_recalculation()
                    return _calc_entry(core, key, func, args, kwds, _print)
                if entry is None or (not entry._completed and not entry._processing):
                    _print("No entry found. No current calc. Calling like a boss.")
                    _mctx.record_miss()
                    _mctx.record_recalculation()
                    return _calc_entry(core, key, func, args, kwds, _print)
                _print("Entry found.")
                if _allow_none or entry.value is not None:
                    _print("Cached result found.")
                    now = datetime.now()
                    max_allowed_age = _stale_after
                    nonneg_max_age = True
                    if max_age is not None:
                        if max_age < ZERO_TIMEDELTA:
                            _print("max_age is negative. Cached result considered stale.")
                            nonneg_max_age = False
                        else:
                            assert max_age is not None  # noqa: S101
                            max_allowed_age = min(_stale_after, max_age)
                    # note: if max_age < 0, we always consider a value stale
                    if nonneg_max_age and (now - entry.time <= max_allowed_age):
                        _print("And it is fresh!")
                        _mctx.record_hit()
                        return entry.value
                    _print("But it is stale... :(")
                    _mctx.record_stale_hit()
                    _mctx.record_miss()
                    if entry._processing:
                        if _next_time:
                            _print("Returning stale.")
                            return entry.value  # return stale val
                        _print("Already calc. Waiting on change.")
                        try:
                            return core.wait_on_entry_calc(key)
                        except RecalculationNeeded:
                            _mctx.record_wait_timeout()
                            _mctx.record_recalculation()
                            return _calc_entry(core, key, func, args, kwds, _print)
                    if _next_time:
                        _print("Async calc and return stale")
                        _mctx.record_recalculation()
                        core.mark_entry_being_calculated(key)

                        def _wrapped_function_thread(
                            core_arg: _BaseCore,
                            key_arg: Any,
                            func_arg: Callable[..., Any],
                            args_arg: tuple[Any, ...],
                            kwds_arg: dict[str, Any],
                        ) -> None:
                            """Run background recalculation and clear processing flag when done."""
                            try:
                                _function_thread(core_arg, key_arg, func_arg, args_arg, kwds_arg)
                            finally:
                                core_arg.mark_entry_not_calculated(key_arg)

                        _get_executor().submit(_wrapped_function_thread, core, key, func, args, kwds)
                        return entry.value
                    _print("Calling decorated function and waiting")
                    _mctx.record_recalculation()
                    return _calc_entry(core, key, func, args, kwds, _print)
                if entry._processing:
                    _print("No value but being calculated. Waiting.")
                    try:
                        return core.wait_on_entry_calc(key)
                    except RecalculationNeeded:
                        _mctx.record_wait_timeout()
                        _mctx.record_miss()
                        _mctx.record_recalculation()
                        return _calc_entry(core, key, func, args, kwds, _print)
                _print("No entry found. No current calc. Calling like a boss.")
                _mctx.record_miss()
                _mctx.record_recalculation()
                return _calc_entry(core, key, func, args, kwds, _print)

        async def _call_async(*args, max_age: Optional[timedelta] = None, **kwds):
            # NOTE: For async functions, wait_for_calc_timeout is not honored.
            # Instead of blocking the event loop waiting for concurrent
            # calculations, async functions will recalculate in parallel.
            # This avoids deadlocks and maintains async efficiency.
            nonlocal allow_none, last_cleanup
            _allow_none = _update_with_defaults(allow_none, "allow_none", kwds)
            # print('Inside async wrapper for {}.'.format(func.__name__))
            ignore_cache = _pop_kwds_with_deprecation(kwds, "ignore_cache", False)
            overwrite_cache = _pop_kwds_with_deprecation(kwds, "overwrite_cache", False)
            verbose = _pop_kwds_with_deprecation(kwds, "verbose_cache", False)
            ignore_cache = kwds.pop("cachier__skip_cache", ignore_cache)
            overwrite_cache = kwds.pop("cachier__overwrite_cache", overwrite_cache)
            verbose = kwds.pop("cachier__verbose", verbose)
            _stale_after = _update_with_defaults(stale_after, "stale_after", kwds)
            _next_time = _update_with_defaults(next_time, "next_time", kwds)
            _cleanup_flag = _update_with_defaults(cleanup_stale, "cleanup_stale", kwds)
            _cleanup_interval_val = _update_with_defaults(cleanup_interval, "cleanup_interval", kwds)
            # merge args expanded as kwargs and the original kwds
            kwargs = _convert_args_kwargs(func, _is_method=core.func_is_method, args=args, kwds=kwds)

            if _cleanup_flag:
                now = datetime.now()
                with cleanup_lock:
                    if now - last_cleanup >= _cleanup_interval_val:
                        last_cleanup = now
                        _get_executor().submit(core.delete_stale_entries, _stale_after)

            _print = print if verbose else lambda x: None

            # Check current global caching state dynamically
            from .config import _global_params

            if ignore_cache or not _global_params.caching_enabled:
                return await func(args[0], **kwargs) if core.func_is_method else await func(**kwargs)  # type: ignore[call-arg,misc]

            with MetricsContext(cache_metrics) as _mctx:
                key, entry = await core.aget_entry((), kwargs)
                if overwrite_cache:
                    _mctx.record_miss()
                    _mctx.record_recalculation()
                    return await _calc_entry_async(core, key, func, args, kwds, _print)
                if entry is None or (not entry._completed and not entry._processing):
                    _print("No entry found. No current calc. Calling like a boss.")
                    _mctx.record_miss()
                    _mctx.record_recalculation()
                    return await _calc_entry_async(core, key, func, args, kwds, _print)
                _print("Entry found.")
                if _allow_none or entry.value is not None:
                    _print("Cached result found.")
                    now = datetime.now()
                    max_allowed_age = _stale_after
                    nonneg_max_age = True
                    if max_age is not None:
                        if max_age < ZERO_TIMEDELTA:
                            _print("max_age is negative. Cached result considered stale.")
                            nonneg_max_age = False
                        else:
                            assert max_age is not None  # noqa: S101
                            max_allowed_age = min(_stale_after, max_age)
                    # note: if max_age < 0, we always consider a value stale
                    if nonneg_max_age and (now - entry.time <= max_allowed_age):
                        _print("And it is fresh!")
                        _mctx.record_hit()
                        return entry.value
                    _print("But it is stale... :(")
                    _mctx.record_stale_hit()
                    _mctx.record_miss()
                    if _next_time:
                        _print("Async calc and return stale")
                        _mctx.record_recalculation()
                        # Mark entry as being calculated; background task will
                        # update cache and clear the flag when done.
                        await core.amark_entry_being_calculated(key)
                        # Use asyncio.create_task for background execution,
                        # ensuring that the processing flag is only cleared
                        # after recomputation completes.
                        asyncio.create_task(_background_recalc_async(core, key, func, args, kwds))
                        return entry.value
                    _print("Calling decorated function and waiting")
                    _mctx.record_recalculation()
                    return await _calc_entry_async(core, key, func, args, kwds, _print)
                if entry._processing:
                    msg = "No value but being calculated. Recalculating"
                    _print(f"{msg} (async - no wait).")
                    # For async, don't wait - just recalculate
                    # This avoids blocking the event loop
                    _mctx.record_miss()
                    _mctx.record_recalculation()
                    return await _calc_entry_async(core, key, func, args, kwds, _print)
                _print("No entry found. No current calc. Calling like a boss.")
                _mctx.record_miss()
                _mctx.record_recalculation()
                return await _calc_entry_async(core, key, func, args, kwds, _print)

        # MAINTAINER NOTE: The main function wrapper is now a standard function
        # that passes *args and **kwargs to _call. This ensures that user
        # arguments are not shifted, and max_age is only settable via keyword
        # argument.
        # For async functions, we create an async wrapper that calls
        # _call_async.
        if is_coroutine:

            @wraps(func)
            async def func_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _R:
                return await _call_async(*args, **kwargs)  # type: ignore[arg-type]

        else:

            @wraps(func)
            def func_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _R:
                return _call(*args, **kwargs)  # type: ignore[arg-type]

        def _clear_cache(*args, **kwds):
            """Clear the cache, or only the entry matching the provided arguments."""
            if args or kwds:
                kwargs = _convert_public_cache_args(func, core.func_is_method, args, kwds)
                key = core.get_key((), kwargs)
                core.clear_cache_entry(key)
            else:
                core.clear_cache()
            if is_coroutine:
                return _ImmediateAwaitable()
            return None

        def _clear_being_calculated():
            """Mark all entries in this cache as not being calculated."""
            core.clear_being_calculated()
            if is_coroutine:
                return _ImmediateAwaitable()
            return None

        async def _aclear_cache(*args, **kwds):
            """Clear the cache asynchronously, or only the entry matching the provided arguments."""
            if args or kwds:
                kwargs = _convert_public_cache_args(func, core.func_is_method, args, kwds)
                key = core.get_key((), kwargs)
                await core.aclear_cache_entry(key)
            else:
                await core.aclear_cache()

        async def _aclear_being_calculated():
            """Mark all entries in this cache as not being calculated asynchronously."""
            await core.aclear_being_calculated()

        def _cache_dpath():
            """Return the path to the cache dir, if exists; None if not."""
            return getattr(core, "cache_dir", None)

        def _precache_value(*args, value_to_cache, **kwds):
            """Add an initial value to the cache.

            Parameters
            ----------
            *args : Any
                Positional arguments used to build the cache key.
            value_to_cache : any
                Entry to be written into the cache.
            **kwds : Any
                Keyword arguments used to build the cache key.

            """
            # merge args expanded as kwargs and the original kwds
            kwargs = _convert_args_kwargs(func, _is_method=core.func_is_method, args=args, kwds=kwds)
            return core.precache_value((), kwargs, value_to_cache)

        func_wrapper.clear_cache = _clear_cache  # type: ignore[attr-defined]
        func_wrapper.clear_being_calculated = _clear_being_calculated  # type: ignore[attr-defined]
        func_wrapper.aclear_cache = _aclear_cache  # type: ignore[attr-defined]
        func_wrapper.aclear_being_calculated = _aclear_being_calculated  # type: ignore[attr-defined]
        func_wrapper.cache_dpath = _cache_dpath  # type: ignore[attr-defined]
        func_wrapper.precache_value = _precache_value  # type: ignore[attr-defined]
        func_wrapper.metrics = cache_metrics  # type: ignore[attr-defined]
        return func_wrapper  # type: ignore[return-value]

    return _cachier_decorator