run any uvicorn based app easily

dbtunnel/__init__.py

@@ -80,5 +80,10 @@ def shiny_python(app, port: int = 8080):
         from dbtunnel.shiny import ShinyPythonAppTunnel
         return ShinyPythonAppTunnel(app, port)
 
+    @staticmethod
+    def uvicorn(app, port: int = 8080):
+        from dbtunnel.uvicorn import UvicornAppTunnel
+        return UvicornAppTunnel(app, port)
+
 
 dbtunnel = AppTunnels()

dbtunnel/tunnels.py

@@ -92,7 +92,7 @@ def get_cloud_proxy_settings(cloud: str, org_id: str, cluster_id: str, port: int
 
 Flavor = Literal[
     "gradio", "fastapi", "nicegui", "streamlit", "stable-diffusion-ui", "bokeh", "flask", "dash", "solara",
-    "code-server", "chainlit", "shiny-python"]
+    "code-server", "chainlit", "shiny-python", "uvicorn"]
 
 
 def get_current_username() -> str:

dbtunnel/uvicorn.py

@@ -0,0 +1,44 @@
+from dbtunnel.tunnels import DbTunnel
+
+
+class UvicornAppTunnel(DbTunnel):
+
+    def __init__(self, asgi_app, port: int = 8080):
+        super().__init__(port, flavor="uvicorn")
+        self._asgi_app = asgi_app
+
+    def _imports(self):
+        try:
+            import uvicorn
+            import nest_asyncio
+        except ImportError as e:
+            self._log.info("ImportError: Make sure you have nest_asyncio and uvicorn installed;"
+                           "pip install fastapi nest_asyncio uvicorn")
+            raise e
+
+    def _run(self):
+        self.display()
+        self._log.info("Starting server...")
+        import nest_asyncio
+        import uvicorn
+        nest_asyncio.apply()
+
+        # Start the server
+        async def start():
+            config = uvicorn.Config(
+                self._asgi_app,
+                host="0.0.0.0",
+                port=self._port,
+                root_path=self._proxy_settings.url_base_path.rstrip("/")
+            )
+            server = uvicorn.Server(config)
+            await server.serve()
+
+        # Run the asyncio event loop instead of uvloop to enable re entrance
+        import asyncio
+        self._log.info(f"Use this link to access your uvicorn based app: \n{self._proxy_settings.proxy_url}")
+        asyncio.run(start())
+
+    def _display_url(self):
+        # must end with a "/" for it to not redirect
+        return f'<a href="{self._proxy_settings.proxy_url}">Click to go to {self._flavor} App!</a>'

setup.py

@@ -25,6 +25,10 @@
             "fastapi",
             "uvicorn",
         ],
+        "uvicorn": [
+            # Specify dependencies for building documentation here
+            "uvicorn",
+        ],
         "streamlit": [
             # Specify dependencies for building documentation here
             "streamlit",
@@ -44,12 +48,12 @@
         "flask": [
             "flask",
             "fastapi",
-            "uvicorn", # no websockets
+            "uvicorn",  # no websockets
         ],
         "dash": [
             "dash",
             "fastapi",
-            "uvicorn", # no websockets
+            "uvicorn",  # no websockets
         ],
         "sql": [
             "databricks-sql-connector"