Bart Broere

Freelance Developer

Adding a custom Tornado handler to a streamlit project

27 Mar 2024

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

By default 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 method of starting your project can no longer be used, however. 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
from streamlit.web.server.websocket_headers import _get_websocket_headers


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 some 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 copied from streamlit.web.bootstrap
        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