🥅 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

@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)!

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

@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)!

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 as`dst`it is recommended to set`terminator` to empty string.	`'\n'`

Source code in src/streamlit_extras/capture/__init__.py

@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)!

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 as`dst`it is recommended to set`terminator` 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

@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)!

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)