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("Successfully 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