A long time ago, I've written a sample application in Flask demonstrating real-time charts using SSE (Server-Sent Events). A few weeks back, I've made some improvements and deployed a working demo for everyone to see it in action. You can visit my blog "Creating Real-Time Charts with Flask" in https://ron.sh/creating-real-time-charts-with-flask/ if you want to learn more.
Now, people asked if the same can be implemented in FastAPI.
What is FastAPI?
FastAPI is a modern, fast (high-performance), web framework for building APIs with Python 3.6+ based on standard Python type hints.
FastAPI is a popular Python web framework that was written to be fast and allowing developers to write less code. You can find more information about FastAPI from https://fastapi.tiangolo.com/.
Server-Sent Events (SSE)
Server-Sent Events is another way of pushing updates to a client. The difference between WebSockets and SSE is that a WebSocket is two-way while SSE is a one-way communication. SSE is optimum for pushing notifications to the client since there is no requirement of sending back messages to the server in real-time.
SSE in FastAPI
To implement SSE in a FastAPI server, it is only needed to stream data using a generator and set mimetype to text/event-stream.
# generate_random_data() is an async generator
return StreamingResponse(generate_random_data(request), media_type="text/event-stream")Implementing Real-Time Charts Using SSE
In this example, we will use FastAPI and Chart.js. The code below shows our FastAPI server implementation. generate_random_data() yield values from 0 to 100 and the current timestamp. As for the frontend code using Chart.js, I will be using the same source you can find in https://ron.sh/creating-real-time-charts-with-flask/.
import json
import logging
import random
import sys
from datetime import datetime
from typing import Iterator
import asyncio
from fastapi import FastAPI
from fastapi.responses import HTMLResponse, StreamingResponse
from fastapi.requests import Request
from fastapi.templating import Jinja2Templates
logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)
application = FastAPI()
templates = Jinja2Templates(directory="templates")
random.seed() # Initialize the random number generator
@application.get("/", response_class=HTMLResponse)
async def index(request: Request) -> templates.TemplateResponse:
return templates.TemplateResponse("index.html", {"request": request})
async def generate_random_data(request: Request) -> Iterator[str]:
"""
Generates random value between 0 and 100
:return: String containing current timestamp (YYYY-mm-dd HH:MM:SS) and randomly generated data.
"""
client_ip = request.client.host
logger.info("Client %s connected", client_ip)
while True:
json_data = json.dumps(
{
"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"value": random.random() * 100,
}
)
yield f"data:{json_data}\n\n"
await asyncio.sleep(1)
@application.get("/chart-data")
async def chart_data(request: Request) -> StreamingResponse:
response = StreamingResponse(generate_random_data(request), media_type="text/event-stream")
response.headers["Cache-Control"] = "no-cache"
response.headers["X-Accel-Buffering"] = "no"
return response
Using sse-starlette
Another method is to use EventSourceResponse from sse-starlette. Simply swap it to StreamingResponse and remove the media type.
return EventSourceResponse(generate_random_data(request))The minor difference is that converting the result of the generator to JSON is not needed anymore. Just yield the dict!
async def generate_random_data(request: Request) -> Iterator[str]:
"""
Generates random value between 0 and 100
:return: String containing current timestamp (YYYY-mm-dd HH:MM:SS) and randomly generated data.
"""
client_ip = request.client.host
logger.info("Client %s connected", client_ip)
while True:
yield {
"time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
"value": random.random() * 100,
}
await asyncio.sleep(1)Source code
Sample application is available on Github.
