Skip to content

πŸ₯…Β Β Capture

Submitted by Alexander Martin

Summary

Capture utility extensions for the standard streamlit library

Functions

redirect

Redirect STDOUT and STDERR to streamlit functions.

Source code in src/streamlit_extras/capture/__init__.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
@extra
@contextmanager
def redirect(src: TextIO, dst: Callable, terminator: str = "\n"):
    """Redirect STDOUT and STDERR to streamlit functions."""
    with StringIO() as buffer:

        def new_write(b):
            buffer.write(b + terminator)
            dst(buffer.getvalue())

        # Test if we are actually running in the streamlit script thread before we redirect
        if get_script_run_ctx() is not None:
            old_write = src.write
            try:
                src.write = new_write  # type: ignore
                yield
            finally:
                src.write = old_write  # type: ignore
        else:
            yield

Import:

from streamlit_extras.capture import redirect # (1)!
  1. You should add this to the top of your .py file πŸ› 

stdout

Capture STDOUT and redirect it to a callable `dst`

Args:
    dst (Callable): A function callable with a single string argument. The entire captured contents will be
        passed to this function every time a new string is written. It is designed to be compatible with
        st.empty().* functions as callbacks.
    terminator (str, optional): If a `terminator` is specified, it is added onto each call to stdout.write/print.
        This defaults to a newline which causes them to display on separate lines within an st.empty.write `dst.
        If using this with st.empty.code as `dst` it is recommended to set `terminator` to empty string. Defaults to "

".

Source code in src/streamlit_extras/capture/__init__.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@extra
@contextmanager
def stdout(dst: Callable, terminator: str = "\n"):
    """
    Capture STDOUT and redirect it to a callable `dst`

    Args:
        dst (Callable): A function callable with a single string argument. The entire captured contents will be
            passed to this function every time a new string is written. It is designed to be compatible with
            st.empty().* functions as callbacks.
        terminator (str, optional): If a `terminator` is specified, it is added onto each call to stdout.write/print.
            This defaults to a newline which causes them to display on separate lines within an st.empty.write `dst.
            If using this with st.empty.code as `dst` it is recommended to set `terminator` to empty string. Defaults to "\n".
    """
    with redirect(sys.stdout, dst, terminator):
        yield

Import:

from streamlit_extras.capture import stdout # (1)!
  1. You should add this to the top of your .py file πŸ› 

stderr

Capture STDERR and redirect it to a callable dst.

Parameters:

Name Type Description Default
dst callable[str]

A funciton callable with a single string argument. The entire captured contents will be passed to this function every time a new string is written. It is designed to be compatible with st.empty().* functions as callbacks.

 required
terminator (optional, str)

If a terminator is specified, it is added onto each call to stdout.write/print. This defaults to a newline which causes them to display on separate lines within an st.empty.write dst. If using this with st.empty.code asdstit is recommended to setterminator` to empty string.

 '\n'
Source code in src/streamlit_extras/capture/__init__.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@extra
@contextmanager
def stderr(dst: Callable, terminator="\n"):
    """
    Capture STDERR and redirect it to a callable `dst`.

    Args:
        dst (callable[str]): A funciton callable with a single string argument. The entire captured contents will be
            passed to this function every time a new string is written. It is designed to be compatible with
            st.empty().* functions as callbacks.
        terminator (optional, str): If a `terminator` is specified, it is added onto each call to stdout.write/print.
            This defaults to a newline which causes them to display on separate lines within an st.empty.write `dst.
            If using this with st.empty.code as `dst` it is recommended to set `terminator` to empty string.
    """
    with redirect(sys.stderr, dst, terminator):
        yield

Import:

from streamlit_extras.capture import stderr # (1)!
  1. You should add this to the top of your .py file πŸ› 

logcapture

Redirect logging to a streamlit function call dst.

Parameters:

Name Type Description Default
dst callable[str]

A function callable with a single string argument. The entire log contents will be passed to this function every time a log is written. It is designed to be compatible with st.empty().* functions as callbacks.

 required
terminator (optional, str)

If a terminator is specified, it is added onto the end of each log. This defaults to a newline which causes them to display on separate lines within an st.empty.write dst. If using this with st.empty.code asdstit is recommended to setterminator` to empty string.

 '\n'
from_logger (optional, Logger or logger)

The logger from which logs will be captured. Defaults to logging.root.

 None
formatter (optional, Formatter)

If specified, the specified formatter will be added to the logging handler to control how logs are displayed.

 None
Source code in src/streamlit_extras/capture/__init__.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
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
@extra
@contextmanager
def logcapture(
    dst: Callable,
    terminator: str = "\n",
    from_logger: logging.Logger | None = None,
    formatter: logging.Formatter | None = None,
):
    """
    Redirect logging to a streamlit function call `dst`.

    Args:
        dst (callable[str]): A function callable with a single string argument. The entire log contents will be
            passed to this function every time a log is written. It is designed to be compatible with st.empty().*
            functions as callbacks.
        terminator (optional, str): If a `terminator` is specified, it is added onto the end of each log.
            This defaults to a newline which causes them to display on separate lines within an st.empty.write `dst.
            If using this with st.empty.code as `dst` it is recommended to set `terminator` to empty string.
        from_logger (optional, logging.Logger or loguru.logger): The logger from which logs will be captured.
            Defaults to `logging.root`.
        formatter (optional, logging.Formatter): If specified, the specified formatter will be added to the logging
            handler to control how logs are displayed.
    """

    if not from_logger:
        from_logger = logging.getLogger()  # root logger

    # Special-case loguru
    using_loguru = (
        "loguru" in sys.modules and sys.modules["loguru"].logger is from_logger
    )

    with StringIO() as buffer:
        new_handler = StreamlitLoggingHandler(buffer)
        new_handler.set_callback(dst)
        new_handler.terminator = terminator
        if formatter:
            new_handler.setFormatter(formatter)
        elif using_loguru:
            pass
        else:
            new_handler.setFormatter(
                logging.Formatter(
                    "%(asctime)s - %(levelname)s %(message)s",
                    datefmt="%m/%d/%Y %I:%M:%S %p",
                )
            )
        handler_id = None
        if using_loguru:
            handler_id = from_logger.add(new_handler)  # type: ignore
        else:
            from_logger.addHandler(new_handler)
        try:
            yield
        finally:
            if using_loguru:
                from_logger.remove(handler_id)  # type: ignore
            else:
                from_logger.removeHandler(new_handler)

Import:

from streamlit_extras.capture import logcapture # (1)!
  1. You should add this to the top of your .py file πŸ› 

Examples

example_stdout

def example_stdout():
    output = st.empty()
    with stdout(output.code, terminator=""):
        print("This is some captured stdout")
        print("How about that, Isn't it great?")
        if st.button("Click to print more"):
            print("You added another line!")
Output (beta)

example_stderr

def example_stderr():
    output = st.empty()
    with stderr(output.code, terminator=""):
        print("This is some captured stderr", file=sys.stderr)
        print(
            "For this example, though, there aren't any problems...yet", file=sys.stderr
        )
        if st.button("Throw an error!"):
            print("ERROR: Task failed successfully", file=sys.stderr)
            print("Psst....stdout isn't captured here")
Output (beta)

example_logcapture

def example_logcapture():
    logger = logging.getLogger("examplelogger")
    logger.setLevel("DEBUG")
    with logcapture(st.empty().code, from_logger=logger):
        logger.error("Roses are red")
        logger.info("Violets are blue")
        logger.warning("This warning is yellow")
        logger.debug("Your code is broke, too")
Output (beta)