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

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