Streamlit Async
Streamlit runs Python scripts top-to-bottom when ever a user interacts with widget.
Streamlit is synchronous by default, meaning each function waits for the previous one to finish.
The problem arises when we are fetching remote APIs, databases or long computations, synchronous code blocks the UI, making apps slow.
This is where asyncio comes in.
Need a Streamlit developer? Click Here
The Execution Model of Streamlit
Streamlit re-runs the entire script when widgets change, this means slow APIs blocks everything.
we use Async to allow multiple tasks to run concurrently, thereby improving responsiveness.
Here the UI freezes for 5 seconds before showing the success message
import time
import streamlit as st
st.title("Synchronous Example")
record_time = time.sleep(5)
def fetch_data():
record_time
return "Data fetched"
if st.button("Fetch Data"):
result = fetch_data()
st.success(result)
The Significance of Async in Streamlit
Without Async, API calls, database queries, or computations block the main thread. The user experience suffers
With Async, multiple API/database calls can run in parallel, UI stays responsive and it’s perfect for real-time dashboards
Streamlit async + await
import asyncio
import streamlit as st
st.title("Async Demo")
async def fetch_data():
await asyncio.sleep(3) #Here we stimulate an API delay
return "Async Data Loaded!"
if st.button("Fetch Data"):
result = asyncio.run(fetch_data())
st.success(result)
Streamlit + Async Using External APIs
Let’s use an aiohttp (non-blocking HTTP client) to retrieve data from an API.
Here, the API does not block the UI
import aiohttp #ensure this is installed using pip install aiohttp
import asyncio
import streamlit as st
st.title("Async API Fetch")
async def get_single_post():
url = "https://jsonplaceholder.typicode.com/posts/1"
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.json()
if st.button("Fetch Post"):
data = asyncio.run(get_single_post())
st.json(data)
Parallelizing Several API Calls.
we can fetch multiple endpoints at once.
Here, instead of waiting for 3 API calls sequentially, All are fetched at once
import aiohttp
import asyncio
import streamlit as st
st.title("Parallel API Requests")
async def fetch_url(session, url):
async with session.get(url) as response:
return await response.json()
async def fetch_all():
urls = [
"https://jsonplaceholder.typicode.com/posts/1",
"https://jsonplaceholder.typicode.com/posts/2",
"https://jsonplaceholder.typicode.com/posts/3"
]
async with aiohttp.ClientSession() as session:
tasks = [fetch_url(session, url) for url in urls]
return await asyncio.gather(*tasks)
if st.button("Fetch All"):
results = asyncio.run(fetch_all())
st.json(results)
Async + Databases in Streamlit
When querying databases (PostgreSQL, Supabase, MongoDB), async helps improve performance.
Here, i connected with my local database. ensure you create one using postgres for this to work.
you can select all from whatever name your database table is
import asyncio
import asyncpg #ensure this is installed using pip install asyncpg
import streamlit as st
st.title("Async Database Query")
async def fetch_users():
conn = await asyncpg.connect("postgresql://postgres:postgres@localhost:5433/postgres")
rows = await conn.fetch("SELECT * From cars LIMIT 5")
await conn.close()
return rows
if st.button("Get Cars"):
cars = asyncio.run(fetch_users())
st.table(cars)
Handling Errors
Here, we use the try and except to display errors or result’s depending on what the case may be.
import aiohttp
import asyncio
import streamlit as st
async def fetch_data():
try:
async with aiohttp.ClientSession() as session:
async with session.get("https://error-url.com") as res:
return await res.json()
except Exception as e:
return {"error": str(e)}
st.title("Async Error Handling")
if st.button("Fetch Data"):
result = asyncio.run(fetch_data())
st.json(result)
To execute jobs concurrently, use asyncio.gather().
Steer clear of asyncio. execute() repeatedly — encapsulate several awaits in a single function.
Make use of st.cache_data to cache answers.
For APIs, use aiohttp; for PostgreSQL, use asyncpg.
Learn more about Streamlit: Click here
Watch Videos on Streamlit:
