Bart Broere

Freelance Developer

Adding a custom Tornado handler to a streamlit project

27 Mar 2024

Update 2024-08-25: The code snippet and information below is now outdated if you’re using the latest version of Streamlit. It can be fixed by checking bootstrap.py in the Streamlit source code and changing your if __name__ == '__main__': block accordingly.

Streamlit uses the Tornado web framework under the hood. All traffic generated by Streamlit originates from Tornado handlers.

Streamlit doesn’t expose much of the Tornado API. In this post I’ll show how you can use it anyway, to add custom handlers, while still enjoying most of the conveniences provided by Streamlit.

The streamlit run start is replaced by code that starts the Tornado server. I build on top of my experiences in this previous post: Adding “a main” to a streamlit dashboard

By subclassing Streamlit’s default Server class, we can modify the routes just before we start the Tornado application. After running the typical setup (Server._create_app()), we add a new routing rule. Since this is appended to the end, and the rule before is set so it matches everything, we need to reverse the order the rules are checked. First the newly added specific rule should be checked, and only after that the default Streamlit routes.

import asyncio

import streamlit.web.bootstrap
from streamlit import config
from streamlit.web.server import Server
from streamlit.web.server.media_file_handler import MediaFileHandler
from streamlit.web.server.server import start_listening
from streamlit.web.server.server_util import make_url_path_regex


streamlit.markdown("# Contents of the streamlit app go here as usual")


class CustomHandler(MediaFileHandler):
    def get_content(self, abspath, start=None, end=None):
        # Implement a custom handler here
        return b''


class CustomServer(Server):
    async def start(self):
        # Override the start of the Tornado server, so we can add custom handlers
        app = self._create_app()

        # Add a new handler
        app.default_router.add_rules([(
                make_url_path_regex(config.get_option("server.baseUrlPath"),
                                    f"custom/(.*)"),
                CustomHandler,
                {"path": ""},
            ),
        ])

        # Our new rules go before the rule matching everything, reverse the list
        app.default_router.rules = list(reversed(app.default_router.rules))

        start_listening(app)
        await self._runtime.start()


if __name__ == '__main__':
    if '__streamlitmagic__' not in locals():
        # Code adapted from bootstrap.py in streamlit
        streamlit.web.bootstrap._fix_sys_path(__file__)
        streamlit.web.bootstrap._fix_tornado_crash()
        streamlit.web.bootstrap._fix_sys_argv(__file__, [])
        streamlit.web.bootstrap._fix_pydeck_mapbox_api_warning()
        streamlit.web.bootstrap._fix_pydantic_duplicate_validators_error()
        streamlit.web.bootstrap._install_pages_watcher(__file__)

        server = CustomServer(__file__, is_hello=False)

        async def run_server():
            await server.start()
            streamlit.web.bootstrap._on_server_start(server)
            streamlit.web.bootstrap._set_up_signal_handler(server)
            await server.stopped

        asyncio.run(run_server())

There’s also a way to replace the default Streamlit routes. In a next post I’ll show how to do that, to prevent unauthorized access to the media assets served by your app.

Related Posts