Skip to content

This module provides utilities to incorporate a standard python logger within streamlit.

Streamlit log handler

StreamlitLogHandler

Bases: Handler

Custom Streamlit log handler to display logs in a Streamlit container

A custom logging handler for Streamlit applications that displays log messages in a Streamlit container.

Attributes:

Name Type Description
container DeltaGenerator

The Streamlit container where log messages will be displayed.
debug bool

A flag to indicate whether to display debug messages.
ansi_escape Pattern

A compiled regular expression to remove ANSI escape sequences from log messages.
log_area DeltaGenerator

An empty Streamlit container for log output.
buffer deque

A deque buffer to store log messages with a maximum length.
_n int

A counter to keep track of the number of log messages seen.

Methods:

Name Description
__init__

Initializes the StreamlitLogHandler with a Streamlit container, buffer length, and debug flag.
n_elems

Returns a string with the total number of elements seen and the number of elements in the buffer. If verb is True, returns a verbose string; otherwise, returns a concise string.
emit

Processes a log record, formats it, appends it to the buffer, and displays it in the Streamlit container. Strips ANSI escape sequences from the log message if present.
clear_logs

Clears the log messages from the Streamlit container and the buffer.
Source code in src/st_logs.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
class StreamlitLogHandler(logging.Handler):
    """
    Custom Streamlit log handler to display logs in a Streamlit container

    A custom logging handler for Streamlit applications that displays log
    messages in a Streamlit container.

    Attributes:
        container (streamlit.DeltaGenerator): The Streamlit container where log messages will be displayed.
        debug (bool): A flag to indicate whether to display debug messages.
        ansi_escape (re.Pattern): A compiled regular expression to remove ANSI escape sequences from log messages.
        log_area (streamlit.DeltaGenerator): An empty Streamlit container for log output.
        buffer (collections.deque): A deque buffer to store log messages with a maximum length.
        _n (int): A counter to keep track of the number of log messages seen.

    Methods:
        __init__(container, maxlen=15, debug=False):
            Initializes the StreamlitLogHandler with a Streamlit container, buffer length, and debug flag.
        n_elems(verb=False):
            Returns a string with the total number of elements seen and the number of elements in the buffer.
            If verb is True, returns a verbose string; otherwise, returns a concise string.
        emit(record):
            Processes a log record, formats it, appends it to the buffer, and displays it in the Streamlit container.
            Strips ANSI escape sequences from the log message if present.
        clear_logs():
            Clears the log messages from the Streamlit container and the buffer.
    """
    # Initialize a custom log handler with a Streamlit container for displaying logs
    def __init__(self, container, maxlen:int=15, debug:bool=False):
        #TODO: find the type for streamlit generic containers
        super().__init__()
        # Store the Streamlit container for log output
        self.container = container
        self.debug = debug
        self.ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])') # Regex to remove ANSI codes
        self.log_area = self.container.empty() # Prepare an empty conatiner for log output

        self.buffer = deque(maxlen=maxlen)
        self._n = 0

    def n_elems(self, verb:bool=False) -> str:
        """
        Return a string with the number of elements seen and the number of elements in the buffer.

        Args:
            verb (bool): If True, returns a verbose string. Defaults to False.

        Returns:
            str: A string representing the total number of elements seen and the number of elements in the buffer.
        """
        if verb:
            return f"total: {self._n}|| in buffer:{len(self.buffer)}"

        return f"{self._n}||{len(self.buffer)}"

    def emit(self, record) -> None:
        '''put the record into buffer so it gets displayed

        Args:
            record (logging.LogRecord): The log record to process and display.

        '''
        self._n += 1
        msg = f"[{self._n}]" + self.format(record)
        self.buffer.append(msg)
        clean_msg = self.ansi_escape.sub('', msg)  # Strip ANSI codes
        if self.debug:
            self.log_area.markdown(clean_msg)

    def clear_logs(self) -> None:
        """
        Clears the log area and buffer.

        This method empties the log area to remove any previous logs and clears the buffer to reset the log storage.
        """
        self.log_area.empty()  # Clear previous logs
        self.buffer.clear()

clear_logs()

Clears the log area and buffer.

This method empties the log area to remove any previous logs and clears the buffer to reset the log storage.

Source code in src/st_logs.py 
 94
 95
 96
 97
 98
 99
100
101
def clear_logs(self) -> None:
    """
    Clears the log area and buffer.

    This method empties the log area to remove any previous logs and clears the buffer to reset the log storage.
    """
    self.log_area.empty()  # Clear previous logs
    self.buffer.clear()

emit(record)

put the record into buffer so it gets displayed

Parameters:

Name Type Description Default
record LogRecord

The log record to process and display.

 required
Source code in src/st_logs.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
def emit(self, record) -> None:
    '''put the record into buffer so it gets displayed

    Args:
        record (logging.LogRecord): The log record to process and display.

    '''
    self._n += 1
    msg = f"[{self._n}]" + self.format(record)
    self.buffer.append(msg)
    clean_msg = self.ansi_escape.sub('', msg)  # Strip ANSI codes
    if self.debug:
        self.log_area.markdown(clean_msg)

n_elems(verb=False)

Return a string with the number of elements seen and the number of elements in the buffer.

Parameters:

Name Type Description Default
verb bool

If True, returns a verbose string. Defaults to False.

 False

Returns:

Name Type Description
str str

A string representing the total number of elements seen and the number of elements in the buffer.
Source code in src/st_logs.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def n_elems(self, verb:bool=False) -> str:
    """
    Return a string with the number of elements seen and the number of elements in the buffer.

    Args:
        verb (bool): If True, returns a verbose string. Defaults to False.

    Returns:
        str: A string representing the total number of elements seen and the number of elements in the buffer.
    """
    if verb:
        return f"total: {self._n}|| in buffer:{len(self.buffer)}"

    return f"{self._n}||{len(self.buffer)}"

demo_log_callback()

basic demo of adding log entries as a callback function

Source code in src/st_logs.py 
175
176
177
178
179
180
181
182
183
184
def demo_log_callback() -> None:
    '''basic demo of adding log entries as a callback function'''

    logger = logging.getLogger(__name__)
    logger.setLevel(logging.DEBUG)
    logger.debug("debug message")
    logger.info("info message")
    logger.warning("warning message")
    logger.error("error message")
    logger.critical("critical message")

parse_log_buffer(log_contents)

Convert log buffer to a list of dictionaries for use with a streamlit datatable.

Parameters:

Name Type Description Default
log_contents deque

A deque containing log lines as strings.

 required

Returns:

Name Type Description
list List[dict]

A list of dictionaries, each representing a parsed log entry with the following keys: - 'timestamp' (datetime): The timestamp of the log entry. - 'n' (str): The log entry number. - 'level' (str): The log level (e.g., INFO, ERROR). - 'module' (str): The name of the module. - 'func' (str): The name of the function. - 'message' (str): The log message.
Source code in src/st_logs.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
def parse_log_buffer(log_contents: deque) -> List[dict]:
    """
    Convert log buffer to a list of dictionaries for use with a streamlit datatable.

    Args:
        log_contents (deque): A deque containing log lines as strings.

    Returns:
        list: A list of dictionaries, each representing a parsed log entry with the following keys:
            - 'timestamp' (datetime): The timestamp of the log entry.
            - 'n' (str): The log entry number.
            - 'level' (str): The log level (e.g., INFO, ERROR).
            - 'module' (str): The name of the module.
            - 'func' (str): The name of the function.
            - 'message' (str): The log message.
    """

    j = 0
    records = []
    for line in log_contents:
        if line:  # Skip empty lines
            j+=1
            try:
                # regex to parsse log lines, with an example line:
                # '[1]2024-11-09 11:19:06,688 - task - run - INFO - 🏃 Running task '
                match = log_pattern.match(line)
                if match:
                    n, timestamp_str, name, func_name, level, message = match.groups()

                # Convert timestamp string to datetime
                timestamp = datetime.strptime(timestamp_str, '%Y-%m-%d %H:%M:%S,%f')

                records.append({
                    'timestamp': timestamp,
                    'n': n,
                    'level': level,
                    'module': name, 
                    'func': func_name,
                    'message': message
                })
            except Exception as e:
                print(f"Failed to parse line: {line}")
                print(f"Error: {e}")
                continue
    return records

setup_logging(level=logging.INFO, buffer_len=15)

Set up logging for the application using Streamlit's container for log display.

Parameters:

Name Type Description Default
level int

The logging level (e.g., logging.INFO, logging.DEBUG). Default is logging.INFO.

 INFO
buffer_len int

The maximum number of log messages to display in the Streamlit container. Default is 15.

 15

Returns:

Name Type Description
StreamlitLogHandler StreamlitLogHandler

The handler that has been added to the root logger.
Source code in src/st_logs.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
@st.cache_resource
def setup_logging(level:int=logging.INFO, buffer_len:int=15) -> StreamlitLogHandler:
    """
    Set up logging for the application using Streamlit's container for log display.

    Args:
        level (int): The logging level (e.g., logging.INFO, logging.DEBUG). Default is logging.INFO.
        buffer_len (int): The maximum number of log messages to display in the Streamlit container. Default is 15.

    Returns:
        StreamlitLogHandler: The handler that has been added to the root logger.
    """
    root_logger = logging.getLogger() # Get the root logger
    log_container = st.container() # Create a container within which we display logs
    handler = StreamlitLogHandler(log_container, maxlen=buffer_len)
    handler.setLevel(level)

    formatter = logging.Formatter('%(asctime)s - %(name)s - %(funcName)s - %(levelname)s - %(message)s')
    handler.setFormatter(formatter)
    root_logger.addHandler(handler)

    #if 'handler' not in st.session_state:
    #    st.session_state['handler'] = handler
    return handler