Check if running in a streamlit session

🎈 Using Streamlit
1

Hi, I want to be able to test various functions I use in my streamlit app through channels like ipython or pytest. However when I run functions that have the @st.cache() decorator they break. So is there a nice way to check if streamlit is running and disable those functions?

Ideally there should be a way to ignore streamlit calls when they do not make sense so that you can run the script in another environment without modification.

For the caching problem I’m thinking something like this:

def cache():
    """Wrapper to disable streamlit caching when running with ipython
    """
    def wrapper(function):
        if st.running:
            wrapped = st.cache(persist=True)(function)
        else:
            wrapped = function
        return wrapped

return wrapper
1 Like
2

Hi @Anderssorby

I am also actively thinking about this.

Currently the answer to me is to split the functions into two.

I think I would start by implementing the general functions in seperate files, modules and or packages. They are of general use and should be implemented as such. I can then work with them and test them in the usual way.

Then in my Streamlit app I wrap those functions with the @st.cache annotation.

from my_module import my_func

@st.cache
def my_func_cached(x):
    return my_func(x)

I would not do the above for small apps and/ or when I start out developing. But as my app and functionality grows, I believe I would then refactor to the above.

Comments and suggestions to this is really appreciated. Maybe I should start writing something about my understanding of Streamlit Architecture Best Practices at https://awesome-streamlit.readthedocs.io/en/latest/. Then hopefully I could get some feedback and improve.

3

And an additional thought is that

  • very often my @st.cache annotated functions are returning dataframes.
  • And those I often will be mutating by creating, updating or deleting columns.
  • If I don’t copy the dataframes from the cache I will be getting errors.

I think I would be developing an annotation to automatically do that deep copy. So maybe ending up with

@copy(deep=True)
@st.cache
def my_func_cached(x):
    return my_func(x)
4

And maybe somehow the @st.cache annotation could be applied in some simpler way?

I’m just brainstorming and this won’t work but could I don’t something along the lines of

from my_module1 import my_func1
from my_module2 import my_func2

my_func1, my_func2 = st.cache(my_func1, my_func2)

Any comments or suggestions?

@thiago?

5

Something like that seems nice.

I ended up writing my own caching decorator that uses numpy memmaps in my situation and that worked nice for my case.

6

Hi all

You can ignore all calls that print to a Streamlit app by setting this config option:

st.set_config('client.displayEnabled', False)

But this doesn’t impact st.cache. Whether this is the desired behavior, it depends on your use-case. We’re open to changing this behavior / adding a separate config option to address other use-cases, but we’d need to understand the reasoning a bit more first.

If you just need to skip the cache for testing purposes, the preferred solution is to patch st.cache like this:

from unittest.mock import patch

def mock_cache(func=None, *args, **kwargs):
  if func is None:
    return lambda f: f
  return func

@patch('streamlit.cache)
def test(cache_mock):
    cache_mock.side_effect = mock_cache

    # The rest of the test goes here.

(Note: I haven’t tested this code!)

As for the ipython use-case, I think it depends on what you want exactly. One hack would be to just overwrite st.cache when in a REPL:

import streamlit as st

if st._util.is_repl():
  st.cache = mock_cache  # see mock_cache definition above

There are other possible solutions too, such as checking whether you’re running in a Streamlit thread first, with:

import streamlit.ReportThread as ReportThread
is_streamlit_thread = ReportThread.get_report_ctx()

if is_streamlit_thread:
  # overwrite st.cache here as before...
3 Likes