Clear cache with a time interval (at least once per hour)?

Here is a cached function, at_time is the timestamp used to control caching.

@st.cache
def get_phrase_data(start_date, end_date, phrases=None, at_time=None,
                             **kwargs):
    """Wrapper for :func:`load_metadata_for_period` for Streamlit caching
    
    start_date and end_date should be ISO format strings in UTC for date
    
    """
    #print('get_phrase_data', phrases)
    data = load_metadata_for_period(start_date, end_date, phrases=phrases,
                    **kwargs)
    return data

And a function to generate the timestamp value for the interval desired

import datetime
def get_truncated_timestamp(end=None, refresh_interval=5, 
                refresh_interval_units='minutes'):
    """A time truncated down to resolution to control caching

    Parameters:
        end (datetime.datetime):  The time to validate against

    Returns:
        datetime.datetime: a timestamp set to the nearest refresh_interval.

    If end is at or before midnight at the start of the
    current day returns **None** as we'll never want to change that
    cache anyway.

    """
    if end and end <= datetime.date.today():
        #  Fix timestamp for data not updating...
        timestamp = None
    else:
        timestamp = datetime.datetime.now()
        if refresh_interval_units == 'minutes':
            interval = int(refresh_interval)
            timestamp = timestamp.replace(
                            minute=(timestamp.minute // interval) * interval,
                            second=0, microsecond=0, )
    return timestamp

And here is some psuedo code to call it.

    data = get_phrase_data(start, end, phrases, get_truncated_timestamp(end),
                              **kwargs)
3 Likes