Useful Streamlit Code Snippets

File Picker

A simple file picker can be made in streamlit by combining the back-end and streamlit's widgets

import streamlit as st
from octostar.utils.ontology import query_ontology

def search_files_with_extension(workspaces, extension):
    workspaces_condition = ("`os_workspace` IN (" + ",".join(["'" + workspace + "'" for workspace in workspaces]) + ")") if workspaces else "FALSE"
    files = query_ontology.sync("SELECT * FROM `dtimbr`.`os_file` WHERE " + workspaces_condition + f" AND `entity_label` LIKE '%.{extension}'")
    files = {f['os_entity_uid']: f for f in files}
    st.session_state['state']['files'] = files
    return files

def load_file(file_record):
    st.session_state['state']['loaded_file'] = read_file.sync(file_record['os_workspace'], file_record['os_entity_uid'])

if st.session_state['state']['files'] is None:
    search_files_with_extension(st.session_state['input']['workspaces'], 'png')
selected_file = st.selectbox('Found files (please select one',
                             [e['os_entity_uid'] for e in st.session_state['state']['files'].values()],
                             format_func=lambda x: st.session_state['state']['files'][x]['entity_label'])
selected_file = st.session_state['state']['files'][selected_file]
col1, col2, _ = st.columns(3)
with col1:
    st.button("Load File", type='primary', on_click=lambda: load_file(selected_file)
with col2:
    st.button("Reload Files", on_click=lambda: search_files_with_extension(st.session_state['input']['workspaces'], 'png'))
st.image(st.session_state['state']['loaded_file']

Concept Picker

A simple concept picker in streamlit showing the full list of concepts in the ontology

import streamlit as st
from octostar.utils.ontology import fetch_ontology_data

def initialize():
    st.session_state['ontology'] = fetch_ontology_data.sync()
    st.session_state['initialized'] = True
    
def loop():
    concept = st.selectbox("Select a Concept",
                 list(st.session_state['ontology']['concepts'].keys()),
                 format_func=lambda x: " ".join([w.capitalize() for w in x.split("_")]))
    concept = st.session_state['ontology']['concepts'][concept]

File Saver

File saver which allows the user to select a workspace (among the available ones with write permissions), write a path and a filename, and select an extension:

import streamlit as st
from io import BytesIO
from octostar.utils.workspace import write_file

workspace_ids = [workspace_id for workspace_id in st.session_state['input']['workspaces']
                 if st.session_state['input']['workspaces'][workspace_id]['os_permission'] > 1]
if workspace_ids:
    col1, col2, col3 = st.columns([0.3, 0.5, 0.2])
    with col1:
        workspace_id = st.selectbox("Workspace", workspace_ids,
            format_func=lambda x: st.session_state['input']['workspaces'][x]['entity_label'],
            index=0)
        workspace_name = st.session_state['input']['workspaces'][workspace_id]['entity_label']
    with col2:
        filename = st.text_input("Filepath", "My Folder/")
    with col3:
        filename_extension = st.selectbox(
        "file extension", ['.txt'], label_visibility='hidden')
        if st.button("Save Data", type="primary"):
            file = bytes("Hello World!", encoding='utf-8')
            write_file.sync(workspace_id,
                            workspace_name + "/" + filename + filename_extension, 'text/plain',
                            BytesIO(file))
            st.success(f"File {workspace_name}/{filename}/{filename_extension} saved successfully!")
else:
    st.warning("No workspaces with write permission are open.")

Getting the Writable Open Workspaces

The result here is a dictionary keyed by workspace ID and whose values are the workspace records (containing e.g. their labels) as well as the os_permission level, which is a PermissionLevel enum value.

from streamlit_octostar_utils.octostar.permissions import get_workspaces
from streamlit_octostar_utils.octostar.client import impersonating_running_user
from octostar.client import dev_mode
from octostar.utils.workspace.permissions import PermissionLevel
import streamlit as st
import os

@impersonating_running_user()
@dev_mode(os.getenv('OS_DEV_MODE'))
def loop(client):
    if 'workspaces' not in st.session_state:
        st.session_state['workspaces'] = get_workspaces(client, st.session_state['workspaces'])
    writable_workspaces = {k:v for k, v in st.session_state['workspaces'].items() if v.get('os_permission', PermissionLevel.NONE) >= PermissionLevel.WRITE}
    st.write(writable_workspaces)
		
loop()

Filtering a Dataframe

If an app needs to display a dataframe, a set of filtering options can be useful. Below is an example of such code, all done using streamlit's native widgets. The example assumes there is a dataframe in memory and that it is a custom dictionary built from a set of Octostar entities (but the code can easily be applied to other scenarios).

import streamlit as st
import re

filter_by = st.multiselect("Filter entities by:", options=["⭐️ Concept Type", "🔤 Keywords"])
extracted_entities = st.session_state['entities']['concepts']
if "⭐️ Concept Type" in filter_by:
    arrowcol, filtercol = st.columns([0.05, 0.95])
    with arrowcol:
        st.write("↳")
    with filtercol:
        concept_types = list(set([c['type'] for c in st.session_state['entities']['concepts'].values()]))
        concept_types = st.multiselect("Values for ⭐️ Concept Type", concept_types,
                                        format_func=lambda x: " ".join([w.capitalize() for w in x.split("_")]))    
        if concept_types:
            extracted_entities = {k:v for k, v in extracted_entities.items() if v['contents']['type'] in concept_types}
if "🔤 Keywords" in filter_by:
    def _concat_all_concept_info(concept):
        all_words = ""
        all_words += " " + concept['label'] + " " + concept['type']
        for _, attr in concept['attributes'].items():
            all_words += " " + str(attr)
        all_words = all_words.lower()
        return all_words
    arrowcol, filtercol = st.columns([0.05, 0.95])
    with arrowcol:
        st.write("↳")
    with filtercol:
        any_keywords = st.text_input("Words to search for 🔤 Keywords")
        if any_keywords:
            any_keywords = re.sub("[^a-zA-Z0-9]+", " ", any_keywords).lower().split()
            extracted_entities = {k:v for k, v in extracted_entities.items()
                                    if any([w in _concat_all_concept_info(v) for w in any_keywords])}
st.dataframe(extracted_entities)

Stateful Button

Very simple button with toggle-like behavior.

def stateful_button(*args, key=None, **kwargs):
    if key is None:
        raise ValueError("Must pass key")
    if key not in st.session_state:
        st.session_state[key] = False
    if "type" not in kwargs:
        kwargs["type"] = "primary" if st.session_state[key] else "secondary"
    if st.button(*args, **kwargs, key=key + "__internal"):
        st.session_state[key] = not st.session_state[key]
        st.rerun()
    return st.session_state[key]

File Picker​

Concept Picker​

File Saver​

Getting the Writable Open Workspaces​

Filtering a Dataframe​

Stateful Button​