How to use server-sent events (SSE) with FastAPI?

Introduction

Prerequisites

Installing FastAPI

pip install "fastapi[all]"

Installing sse-starlette

pip install sse-starlette
pip install asyncio

Creating a simple hello world endpoint

import asyncio
import uvicorn
from fastapi import FastAPI, Request
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello World"}

Running the uvicorn server

uvicorn main:app --reload
{
"message": "Hello World"
}

Adding SSE support to your FastAPI project

from sse_starlette.sse import EventSourceResponse
STREAM_DELAY = 1  # second
RETRY_TIMEOUT = 15000 # milisecond
@app.get('/stream')
async def message_stream(request: Request):
def new_messages():
# Add logic here to check for new messages
yield 'Hello World'
async def event_generator():
while True:
# If client closes connection, stop sending events
if await request.is_disconnected():
break
# Checks for new messages and return them to client if any
if new_messages():
yield {
"event": "new_message",
"id": "message_id",
"retry": RETRY_TIMEOUT,
"data": "message_content"
}
await asyncio.sleep(STREAM_DELAY) return EventSourceResponse(event_generator())

FastAPI with streaming data and Materialize

What is Materialize?

Streaming data with Materialize

@app.get('/stream')
async def message_stream(request: Request):
def new_messages():
# Check if data in table
results = engine.execute('SELECT count(*) FROM sensors_view_1s')
if results.fetchone()[0] == 0:
return None
else:
return True
async def event_generator():
while True:
# If client was closed the connection
if await request.is_disconnected():
break
# Checks for new messages and return them to client if any
if new_messages():
connection = engine.raw_connection()
with connection.cursor() as cur:
cur.execute("DECLARE c CURSOR FOR TAIL sensors_view_1s")
cur.execute("FETCH ALL c")
for row in cur:
yield row
await asyncio.sleep(MESSAGE_STREAM_DELAY) return EventSourceResponse(event_generator())

Conclusion

--

--

Get the Medium app

A button that says 'Download on the App Store', and if clicked it will lead you to the iOS App store
A button that says 'Get it on, Google Play', and if clicked it will lead you to the Google Play store
Bobby Iliev

I am a professional System Administrator with a demonstrated history of working in the internet industry. I am a Linux lover