Logging and tracing for Streamlit in Snowflake¶

You can capture log messages and trace events of your Streamlit app code as it executes and then analyze the results with SQL, for example, to analyze errors. For more information, see Logging, tracing, and metrics.

Set up an event table¶

To log and trace Streamlit app code, you need to create an event table, associate the event table with an account, and set the log and trace events levels. For example:

-- Create an event table if it doesn't already exist
CREATE EVENT TABLE SAMPLEDATABASE.LOGGING_AND_TRACING.SAMPLE_EVENTS;

-- Associate the event table with the account
ALTER ACCOUNT SET EVENT_TABLE = SAMPLEDATABASE.LOGGING_AND_TRACING.SAMPLE_EVENTS;

-- Set the log level for the database containing your app
ALTER DATABASE STREAMLIT_TEST SET LOG_LEVEL = INFO;

-- Set the trace level for the database containing your app
ALTER DATABASE SAMPLEDATABASE SET TRACE_LEVEL = ON_EVENT;

Copy

For more information, see Event table overview.

Example: Logging from a Streamlit app¶

import logging
import streamlit as st

logger = logging.getLogger("simple_logger")

# Write directly to the app
st.title("Simple Logging Example")

# Get the current credentials
session = st.connection('snowflake').session()

def get_log_messages_query() -> str:
    """
    Get data from the `EVENT TABLE` where the logs were created by this app.
    """
    return """
            SELECT
                TIMESTAMP,
                RECORD:"severity_text"::VARCHAR AS SEVERITY,
                RESOURCE_ATTRIBUTES:"db.user"::VARCHAR AS USER,
                VALUE::VARCHAR AS VALUE
            FROM
                SAMPLE_EVENTS
            WHERE
                SCOPE:"name" = 'simple_logger'
            ORDER BY
                TIMESTAMP DESC;
            """

button = st.button("Log a message")

if button:
    try:
        logger.info("Logging an info message through Stremlit App.")
        st.success('Logged a message')
    except Exception as e:
        logger.error("Logging an error message through Stremlit App: %s",e)
        st.error('Logged an error')

sql = get_log_messages_query()

df = session.sql(sql).to_pandas()

with st.expander("**Show All Messages**"):
     st.dataframe(df, use_container_width=True)

Copy

Example: Emitting traces from a Streamlit app¶

Note

The following example requires installing the snowflake-telemetry-python package. For more information, see Adding support for the telemetry package.

import streamlit as st
import time
import random
from snowflake import telemetry

def sleep_function() -> int:
    """
    Function that sleeps for a random period of time, between one and ten seconds.
    """
    random_time = random.randint(1, 10)
    time.sleep(random_time)
    return random_time

def get_trace_messages_query() -> str:
    """
    Get data from the `EVENT TABLE` where the logs were created by this app.
    """
    return """
            SELECT
                TIMESTAMP,
                RESOURCE_ATTRIBUTES :"db.user" :: VARCHAR AS USER,
                RECORD_TYPE,
                RECORD_ATTRIBUTES
            FROM
                SAMPLE_EVENTS
            WHERE
                RECORD :"name" :: VARCHAR = 'tracing_some_data'
                OR RECORD_ATTRIBUTES :"loggin_demo.tracing" :: VARCHAR = 'begin_span'
            ORDER BY
                TIMESTAMP DESC;
            """

def trace_message() -> None:
    """
    Add a new trace message into the event table.
    """
    execution_time = sleep_function()
    telemetry.set_span_attribute("loggin_demo.tracing", "begin_span")
    telemetry.add_event(
        "tracing_some_data",
        {"function_name": "sleep_function", "execution_time": execution_time},
    )

# Write directly to the app
st.title("Simple Tracing Example")

# Get the current credentials
session = st.connection('snowflake').session()

button = st.button("Add trace event")

if button:
    with st.spinner("Executing function..."):
        trace_message()
        st.toast("Succesfully log a trace message!", icon="✅")

sql = get_trace_messages_query()

df = session.sql(sql).to_pandas()

with st.expander("**Show All Trace Messages**"):
     st.dataframe(df, use_container_width=True)

Copy