Skip to content

Instantly share code, notes, and snippets.

@smartm13
Last active January 11, 2023 16:36
Show Gist options
  • Save smartm13/cd7ccc1f7a9506fecef2f598bf9fd325 to your computer and use it in GitHub Desktop.
Save smartm13/cd7ccc1f7a9506fecef2f598bf9fd325 to your computer and use it in GitHub Desktop.
Decorate this before st.experimental_memo and later check if function has already cached a given set of args+kwargs
import functools
def st_cache_monitor(func):
""" A decorator to handle query_cache=hit/miss utility """
@functools.wraps(func)
def wrapper_func(*args, _querying_cache=None, **kwargs):
""" Wrapper to original func to handle special argument _querying_cache """
if _querying_cache is Ellipsis:
raise LookupError("_querying_cache=`miss`")
return func(*args, **kwargs)
return wrapper_func
def st_cache_monitored(func):
""" A decorator to provide query_cache=hit/miss utility """
def query_cache(*args, **kwargs) -> str:
""" Check if given args+kwargs are cache-hit or cache-miss"""
try:
func(*args, _querying_cache=Ellipsis, **kwargs)
return "hit"
except LookupError as e:
if str(e) == "_querying_cache=`miss`":
return "miss"
# covering really un-expected cases
raise e from None # usually un-reachable
func.query_cache = query_cache
return func
@smartm13
Copy link
Author

smartm13 commented Jul 19, 2022

Description:

  1. Define above st_cache_monitor and st_cache_monitored decorators first.
  2. Later declare your cached function with those 2 decorators.
    For example:
@st_cache_monitored
@st.experimental_memo(...)
@st_cache_monitor
def expensive_func(a, b=None, **extras):
    # returning after heavy time.sleeps
  1. Now you can check if its expensive to call the func without actually calling it. Use function.query_cache.

The decorated function gets an additional callable attribute query_cache which accepts the same arguments as the original function and returns hit if given args are already in cache, else miss.

Example Usage:

if expensive_func.query_cache(4.4, [3,2,1] , kwarg_c=4) == "miss":
    st.text("Your config is not in cache. Press button to have results")

if expensive_func.query_cache(4.4) == "hit":
    st.text("No need to wait, your results are already available.")
    st.write(expensive_func(4.4))  # actual call is only here

@smartm13
Copy link
Author

Here goes a longer version, which also works with st.cache:

class _QueryArg:
    def __init__(self, value=None):
        self.value = value

    def __bool__(self):
        return self.value is not None

    def __reduce__(self):
        return _QueryArg, ()


def st_cache_monitor(func):
    """ A decorator to handle query_cache=hit/miss utility """
    @functools.wraps(func)
    def wrapper_func(*args, _querying_cache=_QueryArg(), **kwargs):
        """ Wrapper to original func to handle special argument _querying_cache """
        if _querying_cache:
            raise LookupError("_querying_cache=`miss`")
        return func(*args, **kwargs)
    return wrapper_func


def st_cache_monitored(func):
    """ A decorator to provide query_cache=hit/miss utility """
    @functools.wraps(func)
    def wrapper_func(*args, **kwargs):
        """ Wrapper to cached func to populate special argument _querying_cache """
        return func(*args, _querying_cache=_QueryArg(), **kwargs)

    def query_cache(*args, **kwargs) -> str:
        """ Check if given args+kwargs are cache-hit or cache-miss"""
        try:
            func(*args, _querying_cache=_QueryArg(True), **kwargs)
            return "hit"
        except LookupError as e:
            if str(e) == "_querying_cache=`miss`":
                return "miss"
            # covering really un-expected cases
            raise e from None  # usually un-reachable
        
    wrapper_func.query_cache = query_cache
    return wrapper_func

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment