Using FastAPI’s WebSockets and Elasticsearch to build a real-time app

Want to get Elastic certified? Find out when the next Elasticsearch Engineer training is running!

Elasticsearch is packed with new features to help you build the best search solutions for your use case. Dive into our sample notebooks to learn more, start a free cloud trial, or try Elastic on your local machine now.

WebSockets is a simultaneous bi-directional communication protocol. The idea is that the client and server can maintain a connection open while sending messages to each other, keeping latency as low as possible. This approach is common in real-time applications like chats, activity notifications, or trading platforms, where latency is key and there is a constant exchange of information.

Imagine that you created a messaging app and want to notify users when they receive a new message. You could poll the server by sending an HTTP request every 5 or 10 seconds until there are new messages, or you could keep a WebSockets connection open and let the server push an event that the client listens for, displaying a notifications badge as soon as the message arrives.

In this scenario, Elasticsearch enables fast and flexible search across datasets, making it ideal for real-time applications where users need immediate results.

In this article, we will create a real-time application using FastAPI's WebSockets feature and Elasticsearch.

Prerequisites

• Python version 3.x
• An Elasticsearch instance (self-hosted or on Elastic Cloud)
• An Elasticsearch API key with write permissions

All the code used for this blog post can be found here.

Use case

To show you how to use WebSockets with FastAPI and Elasticsearch, we’ll use a use case where you, as a store owner, want to notify all your users when a certain query is executed to capture their attention. This simulates real-time engagement in search-driven applications, such as promotional campaigns or product interest alerts.

For this use case, we’ll build an application where a customer can search for products and receive notifications when other users perform searches included on a watchlist.

User A searches for “Kindle,” and user B receives a notification in real time.

Ingesting data

In this section, we’ll create the index mappings and ingest the necessary data using a Python script. You can find the following script in the blog post repository.

Ingest script

Create a new file named ingest_data.py that will contain the Python logic for handling data ingestion.

Install the Elasticsearch library to handle requests to Elasticsearch:

Now import the dependencies and initialize the Elasticsearch client using an API key and an Elasticsearch endpoint URL.

Create a method to set up the index mappings under the name “products”.

Now load the product documents using the bulk API to push them into Elasticsearch. The data will be in an NDJSON file located in the project repository.

Finally, call the methods created.

Run the script using the following command in your terminal.

With that done, let’s continue building the app.

WebSockets app

The application's aesthetic will be simplified to improve readability. The full application repository is available here.

The diagram shows a high-level overview of how the WebSocket app interacts with Elasticsearch and multiple users.

App structure

|-- websockets_elasticsearch_app

|-- inegst_data.py

|-- index.html

|-- main.py

Install and import dependencies

Install FastAPI and WebSocket support. Uvicorn will serve as the local server, Pydantic is used to define data models, and the Elasticsearch client allows the script to connect to the cluster and send data.

FastAPI provides easy-to-use, lightweight, and high-performance tools to build a web application, while Uvicorn serves as the ASGI server to run it. Pydantic is used internally by FastAPI for data validation and parsing, making it easier to define structured data. WebSockets provides the low-level protocol support needed to enable real-time, bidirectional communication between the server and clients. The Elasticsearch Python library was installed earlier and will be used in this application to handle data retrieval.

Now, import the necessary libraries to build the backend.

Elasticsearch client

Define the environment variables for the Elasticsearch endpoint and API key, and instantiate an Elasticsearch client to handle the connection to your Elasticsearch cluster.

Data models and app setup

Now it’s time to create the FastAPI instance, which will handle both REST API and WebSocket routes. Then, we’ll define a few data models using Pydantic.

• The Product model describes what each product looks like.
• The SearchNotification model shapes the message we’ll send to other users.
• The SearchResponse defines how the Elasticsearch results will be returned.

These models help keep things consistent and readable across the app and provide data validation, default values, and autocomplete in our code IDE.

WebSockets endpoint setup

When a user connects to the /ws endpoint, the WebSocket connection is kept open and added to a global list. This allows the server to broadcast messages to all connected clients instantly. If a user disconnects, their connection is removed.

Search endpoint

Now let's review the code where the real-time interaction happens.

When a user performs a search, Elasticsearch is queried and the results are returned. At the same time, if the query is on the global watchlist, all other connected users are notified that someone has found one of those products. The notification includes the query.

The session_id param is used to avoid sending the notification back to the user who initiated the search.

Note: The session_id is based on the current timestamp just for simplicity—in production, you'd want something more robust.

Client-side

To show the application flow, create a frontend using a simple HTML page with a search input, a results area, and a dialog box for notifications.

The notification uses a <dialog> element for demo purposes, but in a real application, you’d likely show a toast or a small badge instead. In real-world scenarios, this type of notification can be used to show how many users are searching for certain products, provide real-time updates on remaining stock, or highlight popular search queries that are returning successful results.

Script tag

Within the <script> tags, include the logic that connects the frontend to the WebSocket endpoint on the backend. Let’s take a look at the code snippet below.

When the page loads, it generates a unique session ID and connects to the WebSocket.

The function connectWebSocket establishes the WebSocket connection using the line `ws = new WebSocket("ws://localhost:8000/ws"). The statement ws.onopen notifies the backend that a new connection has been created. Then, ws.onmessage listens for notifications from other users who are searching for something in the store.

The function showSearchNotification displays the notification received via WebSockets on the screen, while the closeNotification function is used to close the message shown by showSearchNotification.

The searchProducts() function sends the user’s query to the backend and updates the results area with the matching products by calling the displaySearchResults function.

Render the view and main method

Finally, render the HTML page when the app is accessed in the browser and start the server.

Run the application

Run the FastAPI app using uvicorn.

Now the app is live!

Testing the application

Go to localhost:8000/ to render the view of the application and observe what happens in the console:

When a view is opened, the server receives one WebSocket connection. There will be one additional connection for each new page opened. For example, if you open the page in three different browser tabs, you’ll see three connections in the console:

If you close one tab, the corresponding connection is closed:

With multiple active client connections, when one user searches for a product and if the term is on the watchlist, the notification will be received by the other connected clients in real-time.

An optional step is to apply some styles using Tailwind. This improves the UI and gives it a modern, visually appealing look. The full code with the updated UI can be found here.

Conclusion

In this article, we learned how to create real-time notifications based on search using Elasticsearch and FastAPI. We selected a fixed list of products to notify about, but you can explore more custom flows where users can choose their own products or queries to be notified about—or even configure notifications based on product specifications using Elasticsearch's percolate query.

We also experimented with a single pool of users who got notified. With WebSockets, you can choose to broadcast to all users or select specific users. A common pattern is defining “groups” of messages that users can subscribe to, just like group chats.