Hi everyone,
This is a problem we encountered while building a large project at my company. And yes, we are using Streamlit — providing a lot of features such as background tasks with notifications, sidebar navigation (like ERP applications), very fast rendering with skeleton loading, and many more.
We did find a solution to run background tasks with notifications, though I have to mention that the approach is not simple. The entire project has a very complex software architecture because it needs to perform like a rocket. We built a sort of “spider web” of inheritances to make this work with Streamlit.
In the end, we implemented background task execution (on a different node, with FastAPI sending the task) and wrapped a state checker inside a fragment that runs every 20–30 seconds — depending on the refresh interval you want.
This is the only solution we’ve found so far, because if you want to build a push-notification trigger, you need to use COI methods, which are not very stable (I have a theme changer component that uses this method: streamlit-component-theme-changer).
It would be greatly appreciated if Streamlit provided a native method that allows users to send custom messages between the backend and frontend, enabling the creation of more reactive systems.
In the example above, it’s just a test with a counter running inside the fragment, rendering custom HTML with styling to display the information you want — but in a consistent way across all pages. You need to call this fragment on every page (for example, by creating an abstract class and invoking it at the beginning of the runner method).
Here is the example
def common():
with st.container(key="notification"):
st.write(
"""<style>
div:has( > div > .st-key-notification) {
position: absolute;
}
.st-key-notification > div:has(style) {
position: absolute;
visibility: hidden;
}
.st-key-notification:hover, .st-key-notification:has(#notification-icon-check:checked) {
height: 10rem;
}
.st-key-notification {
position: fixed;
right: 1rem;
background: #f8f4f42e;
border-radius: 0.25rem;
display: flex;
justify-content: flex-start;
align-items: center;
flex-direction: column;
height: 2.25rem;
width: 2.10rem;
padding: 0.25rem 0.5rem;
box-sizing: content-box;
backdrop-filter: blur(5px);
box-shadow: 5px 5px 12px #0000002b;
overflow: hidden;
z-index: 1;
transition: height 0.3s ease, width 0.3s ease;
}
.st-key-notification:has(#notification-icon-check:checked) {
width: 15rem;
}
[data-testid="stVerticalBlockBorderWrapper"]:has(> div > .st-key-notification-icon-wrapper) {
width: 100%;
}
.st-key-notification-icon-wrapper {
display: flex;
align-items: flex-end;
width: 100%;
}
.notification-icon {
display: inline-block;
text-align: center;
font-family: "Material Symbols Rounded";
font-weight: 400;
user-select: none;
vertical-align: bottom;
white-space: nowrap;
overflow-wrap: normal;
padding: 0.5rem;
line-height: 1;
border-radius: 0.25rem;
border: 1px solid transparent;
}
.st-key-notification-msg {
width: 100%;
overflow-y: auto;
max-height: 10rem;
}
.notification-icon:hover {
background: #b0b0b036;
}
.notification-icon:has(+ #notification-icon-check:checked) {
background: #b0b0b036;
color: #000;
border: 1px solid #b0b0b036;
}
</style>""",
unsafe_allow_html=True
)
with st.container(key="notification-icon-wrapper"):
st.write(
"""
<label role="img" aria-label="notifications icon" for="notification-icon-check" class="notification-icon">notifications</label>
<input type="checkbox" id="notification-icon-check" style="display: none;">
""",
unsafe_allow_html=True
)
st_box = st.empty()
if st.session_state.has_processing_task:
notification_index(st_box)
else:
with st_box.container(key="notification-msg"):
st.write()
def page1():
common()
st.title("Page1")
if st.button("Start processing task"):
st.session_state.has_processing_task = True
st.session_state.process_percent = 0
st.write("Processing task started")
# Only to trigger the notification_index
st.rerun()
def page2():
common()
st.title("Page2")
def page3():
common()
st.title("Page3")
@st.fragment(run_every=1)
def notification_index(st_box):
print("Processing task...", st.session_state.process_percent)
st.session_state.process_percent += 1
with st_box.container(key="notification-msg"):
st.write(f"Processing task... {st.session_state.process_percent}")
if st.session_state.process_percent > 10:
st.session_state.has_processing_task = False
st.session_state.process_percent = 0
st.rerun()
def run():
if "has_processing_task" not in st.session_state:
st.session_state.has_processing_task = False
if "process_percent" not in st.session_state:
st.session_state.process_percent = 0
page = st.navigation(pages={
"Section 1": [st.Page(page1)],
"Section 2": [st.Page(page2), st.Page(page3)],
})
page.run()
if __name__ == '__main__':
run()
