Session state not working properly

I am having issues with setting the streamlit session state. In the code below, check the last two lines. Just after setting the state I am try to print it but I am getting KeyError (basically that the ā€˜typeā€™ key is not present in the session state). I have spent a good amount of time on this but not sure what the issue is. I am using Streamlit version 1.23.1

Code:

@staticmethod
def add(data, type) -> bool:
   obj = StCache.get(data.uuid, type)
   if obj:
      StCache.update(data, type)
   else:
      if type in st.session_state:
         object_list = st.session_state[type]
      else:
         object_list = []
      
      object_list.append(data)
      st.session_state[type] = object_list
      print(st.session_state[type])

What is StCache?

The issue youā€™re experiencing with the Streamlit session state is likely related to the usage of the variable name type, which is a built-in Python keyword. Using type as a variable name can lead to unexpected behavior and conflicts with the st.session_state dictionary.

To resolve this issue, I recommend choosing a different variable name instead of type. Hereā€™s an updated version of your code with the variable name changed to data_type:

@staticmethod
def add(data, data_type) -> bool:
   obj = StCache.get(data.uuid, data_type)
   if obj:
      StCache.update(data, data_type)
   else:
      if data_type in st.session_state:
         object_list = st.session_state[data_type]
      else:
         object_list = []
      
      object_list.append(data)
      st.session_state[data_type] = object_list
      print(st.session_state[data_type])

By changing the variable name to data_type, you avoid conflicts with the built-in type keyword, which should resolve the KeyError issue you were encountering.

Additionally, itā€™s worth noting that when modifying the st.session_state dictionary, Streamlit expects the keys to be strings. Therefore, you may also need to convert data_type to a string if itā€™s not already.

data_type_str = str(data_type)
if data_type_str in st.session_state:
    object_list = st.session_state[data_type_str]
    ...
    st.session_state[data_type_str] = object_list
    ...
    print(st.session_state[data_type_str])

I hope this helps resolve the issue with the Streamlit session state. Let me know if you have any further questions!

Itā€™s a simple class I have made to manage cache (through session_state). Hereā€™s the complete code:

class StCache:
    @staticmethod
    def get(uuid, type):
        if type in st.session_state:
            for ele in st.session_state[type]:
                if ele['uuid'] == uuid:
                    return ele
        
        return None
    
    @staticmethod
    def update(data, type) -> bool:
        object_found = False

        if type in st.session_state:
            object_list = st.session_state[type]
            for ele in object_list:
                if ele['uuid'] == data.uuid:
                    ele = data
                    object_found = True
                    break
            
            st.session_state[type] = object_list

        return object_found
    
    @staticmethod
    def add(data, data_type) -> bool:
        obj = StCache.get(data.uuid, data_type)
        if obj:
            StCache.update(data, data_type)
        else:
            if data_type in st.session_state:
                object_list = st.session_state[data_type]
            else:
                object_list = []
            
            object_list.append(data)

            st.session_state[data_type] = object_list
            print("first user added: ", st.session_state[data_type])


    @staticmethod
    def delete(uuid, type) -> bool:
        object_found = False
        if type in st.session_state:
            object_list = st.session_state[type]
            for ele in object_list:
                if ele['uuid'] == uuid:
                    object_list.remove(ele)
                    object_found = True
                    break
            
            st.session_state[type] = object_list
        
        return object_found
    
    @staticmethod
    def delete_all(type) -> bool:
        if type in st.session_state:
            del st.session_state[type]
            return True
        
        return False
    
    @staticmethod
    def add_all(data_list, type) -> bool:
        object_list = []
        if type in st.session_state:
            object_list = st.session_state[type]
        
        object_list.extend(data_list)
        st.session_state[type] = object_list

        
        return True
        
    @staticmethod
    def get_all(type):
        if type in st.session_state:
            return st.session_state[type]
        
        return []

tried this but itā€™s not working. I also tried passing it as a string using f"{data_type}" (not much of a point but thought I will give it a go). Does session_state has any scope or does it get cleared upon some action?

You got that error when you call StCache.add()? How are you calling it?

I have a DataRepo for fetching api data or local data like this.

@cache_data
class DataRepo:
    def __init__(self):
        if SERVER != ServerType.PRODUCTION.value:
            self.db_repo = DBRepo()
        else:
            self.db_repo = APIRepo()
    
    def create_user(self, **kwargs):
        user = self.db_repo.create_user(**kwargs).data['data']
        return InternalUserObject(**user) if user else None
    
    def get_first_active_user(self):
        user = self.db_repo.get_first_active_user().data['data']
        return InternalUserObject(**user) if user else None

the DataRepo has a decorator cache_data, which caches API data so that I donā€™t have to hit the database everytime. Inside the cache_data decorator I am calling StCache.add. Code snippet from cache_data

def cache_data(cls):
    
    def _cache_get_first_active_user():
        def _cache_get_first_active_user_wrapper(self, *args, **kwargs):
            logger.log(LoggingType.DEBUG, 'get first user cache', {})
            user_list = StCache.get_all(CacheKey.USER.value)
            if user_list and len(user_list):
                logger.log(LoggingType.DEBUG, '&&&&&& first user found', {'user_list': user_list})
                return user_list[0]
            
            original_func = getattr(cls, '_original_get_first_active_user')
            user = original_func(self, *args, **kwargs)
            if user:
                StCache.add(user, CacheKey.USER.value)
            
            return user
            
        return _cache_get_first_active_user_wrapper
    
    setattr(cls, '_original_get_first_active_user', cls.get_first_active_user)
    setattr(cls, "get_first_active_user", _cache_get_first_active_user())

    return cls

cache_data basically replaces the functions present inside DataRepo with functions having cache logic. In the above code after fetching the user I am saving it in cache using StCache.add

Is there a way to watch session_state, how it changes when the code is run (globally)? From what I am debugging, it works fine the first time but fails every time after that.

I guess you could use a debugger, but I donā€™t use debuggers and I donā€™t know how to use one with streamlit.

What I am trying to do is run your code, see the error happening, and then figure out what exactly is wrong and how to fix it. But there are missing pieces in your code and each time I ask for them you give me more code with even more missing pieces. So every step takes me further away from my goal.

I mostly use debugger but not sure how to it use it to watch session_state. There is definitely a lot more code involved but you can use dummy/placeholder functions in your testing. To better replicate the issue make a function inside main() (which is inside app.py), say test(). and inside ā€˜testā€™ write this code (basically to replicate reloading of the app)

I donā€™t see how st.session_state is different from any other variable in that regard.

Sure, but then I would rather go back to the OP and tell you that your add() function works without errors when called with my dummy parameters and using my dummy StCache, specifically the last two lines in the else part.

st.session_state is showing as an object in my watch list not a variable. other thing, regarding the session_state, I believe the issue is related to threads. On startup streamlit spins up multiple threads, and apart from one the other threads are not able to write in the session_state (in my case) - thatā€™s why I told you to add it inside main(). anyways I am looking, will update this discussion if I find anything.

I am not even sure what you were telling me there.I should write a function main(). I guess I should also make sure that main() is called when the program runs, shouldnā€™t I?.

Then I should define a function test() inside main()? Or call test() from main()? Or both?

And what should I put in test()? All the class and function definitions you posted? Something else?

Dude relax, no need to be passive aggressiveā€¦ use a little common sense, ask if you donā€™t understand and nobody is going to give their complete codebase to you for debugging a problem. If you donā€™t know the solution, have never encountered it or are not able to solve it (or donā€™t have the patience to solve it), just leave it. You are the one who told me that you are trying to run the code (which obviously doesnā€™t make much sense since this was just a small snippet) and therefore I gave some direction on how to replicate it. You could have saved both of our time, by just saying you canā€™t replicate it or canā€™t figure it out. " I donā€™t see how st.session_state is different from any other variable in that regard" - another brilliant insight from you that was of no use.

Well, there are cases where one can spot the cause of the error just by staring at the code. I donā€™t think this is one of them. Hopefully I will be proven wrong.

Solved it by setting fastReruns to false (itā€™s enabled by default)

This topic was automatically closed 2 days after the last reply. New replies are no longer allowed.