Embed Bokeh Plots with FastAPI¶
It's often the case that -- in addition to a standard, data-centric API -- consumers of your service will want a visualization of the data being provided by your service. The goal of this article is to advertise the ease with which such a thing can be achieved in Python, using the bokeh and FastAPI libraries.
First, we'll present a simple application which displays some random data at the URL endpoint /random, based on some path parameters. Then, we implement a second route, /random/plot, which behaves the same way, with the exception that instead we now serve a plot of this random data, rather that the data itself.
If you prefer to jump in and simply read the code for yourself, the example code discussed here can be found in the Github repo here.
Part 1: The FastAPI App¶
To begin, let's introduce the code for the /random route.
NOTE: If you're following along, you'll need to first install
bokeh,FastAPI, anduvicorn(to run the server).
%%capture
%%bash
pip install bokeh FastAPI uvicorn
This route will return an N x 2 array of random numbers between 0 and 1, as seen below:
# %load random_app.py
from random import random
from fastapi import FastAPI, Query
import uvicorn
app = FastAPI(title="Random API")
@app.get("/random", tags=["Random Spot Generator"])
async def get_random_numbers(N: int = Query(default=10, gt=0, le=100)):
return {"spots": [[random(), random()] for i in range(N)]}
def run_server():
uvicorn.run(app)
To run this server, simply type uvicorn fastapi_bokeh:app at the command line to start the server on port 8000 or, alternatively, run the run_server() function. To demo the app, let's do the latter in a separate process. For this we'll need the multiprocess library, and -- later -- the requests module to demo the behaviour of the route itself.
%%capture
%%bash
pip install requests multiprocess
import multiprocess as mp
app_process = mp.Process(target=run_server)
app_process.start()
Once the server is running, random 2D points can then be generated at http://localhost:8000/random, and retrieved using the requests library as seen below:
import requests
requests.get("http://localhost:8000/random", params={"N": 5}).json()
The server's Swagger documentation can be found at http://localhost:8000/docs.
# Displays the contents of a webpage inline in the Jupyter Notebook or -- alternatively --
# as an image, depending on whether IFrame or Image is used
from IPython.display import IFrame, Image
# IFrame("http://localhost:8000/docs", 800, 600)
Image("img/docs1.png", width=800)
Part 2: The Plot¶
Given this nice two-dimensional array of data, we now wish to display it in a bokeh plot. This can be achieved with the following lines of code:
# %load plot.py
from random import random
from bokeh.plotting import figure, curdoc
from bokeh.layouts import column
import requests
def get_spots(N):
try:
response = requests.get("http://localhost:8000/random", params={"N": N}).json()
except Exception as e:
# If the server's not running, using stand-in values instead
response = {"spots": [[random(), random()] for _ in range(N)]}
spots = response["spots"]
return {
"x": [spot[0] for spot in spots],
"y": [spot[1] for spot in spots],
}
def add_figure(doc):
p = figure()
data = get_spots(10)
p.circle("x", "y", source=data)
doc.add_root(column(p))
doc = curdoc()
add_figure(doc)
A quick note about how the data is being generated: in get_spots(N) we attempt to make a call to the API first, and — if it's running — then the retrieved data is used. Otherwise, we populate some stand-in values so that the plot can be viewed even if the service is unavailable.
This plot can be viewed in-browser by running
bokeh serve --show plot.py
at the command line, or in a notebook as shown below:
from bokeh.io import output_notebook, show
output_notebook()
# May be different in your case, depending on where Jupyter is being hosted
NOTEBOOK_PORT = 8890
# from plot import add_figure
show(add_figure, notebook_url="http://localhost:" + str(NOTEBOOK_PORT))
from IPython.display import Image
Image("img/plot.png", width=600)
Part 3: Making the Plot Interactive¶
This is all well and good, but we can do better: since the API endpoint also takes a query parameter $N$, we can do the same here by introducing a slider which will update the number of random spots being plotted when dragged.
# %load plot_slider.py
from random import random
from bokeh.plotting import figure, curdoc
from bokeh.layouts import column
from bokeh.models import Slider, ColumnDataSource
import requests
def get_spots(N):
try:
response = requests.get("http://localhost:8000/random", params={"N": N}).json()
except Exception as e:
# Server not running -- using stand-in values instead
response = {"spots": [[random(), random()] for _ in range(N)]}
spots = response["spots"]
return {
"x": [spot[0] for spot in spots],
"y": [spot[1] for spot in spots],
}
def add_figure(doc):
N = 10
source = ColumnDataSource(data=get_spots(N))
p = figure(x_range=[0,1], y_range=[0,1])
p.circle("x", "y", source=source)
def callback(attr, old, new):
source.data = get_spots(new)
slider = Slider(start=1, end=100, value=N, title="Number of spots")
slider.on_change("value", callback)
doc.add_root(column(p, slider))
doc = curdoc()
add_figure(doc)
This then gives the following behaviour, which can also be viewed in the browser after running bokeh serve --show plot_slider.py:
show(add_figure, notebook_url="http://localhost:" + str(NOTEBOOK_PORT))
from IPython.display import Image
Image("img/plot-slider.png", width=600)
Part 4: Hosting the bokeh plot via FastAPI¶
Now that we have an interactive plot, we'd like to add a second route to our FastAPI app that will serve up this plot alongside the previous route for the data alone. Before we do that, though, let's kill the previous FastAPI app:
app_process.terminate()
app_process.join()
app_process.is_alive()
The code that integrates the bokeh server with the FastAPI app is shown below. Let's take a look, and then discuss the essential ingredients afterwards:
# %load fastapi_bokeh.py
from random import random
from threading import Thread
import asyncio
import logging
from fastapi import FastAPI, Query, Request
from fastapi.templating import Jinja2Templates
from bokeh.server.server import Server
from bokeh.embed import server_document
from tornado.ioloop import IOLoop
import uvicorn
import plot_slider
# Because of https://github.com/tornadoweb/tornado/issues/775
# this code is required to avoid a call to logging.basicConfig() when using tornado
log = logging.getLogger('tornado')
handler = logging.NullHandler()
log.addHandler(handler)
app = FastAPI()
templates = Jinja2Templates(directory='.')
@app.get("/random", tags=["Random Spot Generator"])
async def get_random_numbers(N: int = Query(default=10, gt=0, le=100)):
return {"spots": [[random(), random()] for i in range(N)]}
@app.get("/random/plot", tags=["Random Spot Generator"])
def get_random_numbers_plot(request: Request):
script = server_document("http://localhost:8001/plot-slider")
return templates.TemplateResponse(
"template.html",
{"script": script, "request": request, "framework": "FastAPI"},
)
def start_bokeh_server():
server = Server(
{
"/plot-slider": plot_slider.add_figure,
},
io_loop=IOLoop(),
address="localhost",
port=8001,
allow_websocket_origin=["localhost:8000", "localhost:8001"],
)
server.start()
server.io_loop.start()
def run_server():
bokeh_thread = Thread(target=start_bokeh_server, daemon=True)
bokeh_thread.start()
uvicorn.run(app, host="localhost", port=8000)
As before, the FastAPI app is hosted at localhost port 8000:
uvicorn.run(app, host="localhost", port=8000)
However, before running uvicorn, we start up a separate thread which runs the bokeh server:
bokeh_thread = Thread(target=start_bokeh_server, daemon=True) bokeh_thread.start()
Marking a thread as daemonic means that it will be killed when the main thread exits (which is what we want, so that the bokeh server doesn't continue to run indefinitely in the background).
Within start_bokeh_server(), there is — of course — the bokeh server itself:
server = Server( { "/plot-slider": plot_slider.add_figure, }, io_loop=IOLoop(), address="localhost", port=8001, allow_websocket_origin=["localhost:8000", "localhost:8001"], )
This server will host the route /plot-slider at port 8001 to access the plot and slider widget, using the add_figure(doc) function from Part 3. We also specify that websocket connections be allowed from both ports 8000 and 8001.
We also introduce a templates variable, which loads any Jinja templates from the current directory:
templates = Jinja2Templates(directory='.')
Within the /random/plot route, we return a template populated from template.html which is shown below:
# %load template.html
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<title>{{ framework }} + Bokeh</title>
</head>
<body>
<header>
<h1>Random Numbers</h1>
<p>Use the slider to adjust the number of random points displayed.</p>
</header>
{{ script|safe }}
</body>
</html>
The /random/plot route generates a script to be injected into the template, and returns a populated template:
@app.get("/random/plot", tags=["Random Spot Generator"]) def get_random_numbers_plot(request: Request): script = server_document("http://localhost:8001/plot-slider") return templates.TemplateResponse( "template.html", {"script": script, "request": request, "framework": "FastAPI"}, )
As an example, the framework name is passed into the template, but other values can be passed in equally easily. (The request parameter is required by FastAPI in generating the template.) The script injected is the javascript needed to embed the bokeh plot and slider within our template.
To see this all in action, let's start up another app_process to run this server:
# %load run.py
from fastapi_bokeh import run_server
import multiprocess as mp
app_process = mp.Process(target=run_server)
app_process.start()
As you can see, at http://localhost:8000/random/plot we find the populated template, with the header, plot, and slider:
# Displays the contents of a webpage inline in the Jupyter Notebook or -- alternatively --
# as an image, depending on whether IFrame or Image is used
from IPython.display import IFrame, Image
# IFrame("http://localhost:8000/docs", 800, 800)
Image("img/docs2.png", width=800)
# Displays the contents of a webpage inline in the Jupyter Notebook
from IPython.display import IFrame, Image
# IFrame("http://localhost:8000/random/plot", 800, 800)
Image("img/plot-template.png", width=600)
And, similarly, if we navigate to http://localhost:8001/plot-slider we find the plot alone:
# Displays the contents of a webpage inline in the Jupyter Notebook
from IPython.display import IFrame, Image
# IFrame("http://localhost:8001/plot-slider", 800, 800)
Image("img/plot-slider.png", width=600)
Finally, to clean up, we can kill the app_process:
app_process.terminate()
app_process.join()
app_process.is_alive()
Summary¶
The purpose of this example has been to demonstrate how a bokeh server can be easily adapted to run alongside a FastAPI app, so that rich, interactive plots and widgets can be served to API consumers, in addition to more data-centric routes.
I hope you found this article interesting! Feel free to follow me on Github here