Tutorial (WebSockets)#
In this tutorial, we’re going to build a WebSocket server using Falcon. We’ll start with a simple server that echoes back any message it receives.
We’ll then add more functionality to the server, such as sending JSON data and logging messages.
Note
This tutorial covers the asynchronous flavor of Falcon using the ASGI protocol.
A Falcon WebSocket server builds upon the ASGI WebSocket specification. Therefore it’s not supported in a Falcon WSGI application.
First Steps#
We’ll start with a clean working directory and create a new virtual environment
using the venv
module:
$ mkdir ws_tutorial
$ cd ws_tutorial
$ python3 -m venv .venv
$ source .venv/bin/activate
Create the following directory structure:
ws_tutorial
├── .venv
└── ws_tutorial
├── __init__.py
└── app.py
And next we’ll install Falcon and Uvicorn in our freshly created virtual environment:
$ pip install falcon uvicorn
Now, let’s create a simple Falcon application to ensure our project is working as expected.
import falcon.asgi
import uvicorn
app = falcon.asgi.App()
class HelloWorldResource:
async def on_get(self, req, resp):
resp.media = {'hello': 'world'}
app.add_route('/hello', HelloWorldResource())
if __name__ == '__main__':
uvicorn.run(app, host='localhost', port=8000)
Now we can test the application with httpie
(installable with
pip install httpie
) by running the following command:
$ http localhost:8000/hello
HTTP/1.1 200 OK
content-length: 18
content-type: application/json
date: Sat, 13 Jul 2024 09:13:24 GMT
server: uvicorn
{
"hello": "world"
}
Awesome, it works! Now let’s move on to building our WebSocket server.
WebSockets Server#
We will update our server to include a websocket route that will echo back any message it receives. Later we’ll update the server with more logic, but for now, let’s keep it simple.
import falcon.asgi
from falcon import WebSocketDisconnected
from falcon.asgi import Request, WebSocket
import uvicorn
app = falcon.asgi.App()
class HelloWorldResource:
async def on_get(self, req, resp):
resp.media = {'hello': 'world'}
class EchoWebSocketResource:
async def on_websocket(self, req: Request, ws: WebSocket):
try:
await ws.accept()
except WebSocketDisconnected:
return
while True:
try:
message = await ws.receive_text()
await ws.send_text(f"Received the following text: {message}")
except WebSocketDisconnected:
return
app.add_route('/hello', HelloWorldResource())
app.add_route('/echo', EchoWebSocketResource())
if __name__ == '__main__':
uvicorn.run(app, host='localhost', port=8000)
We’ll also need to install a websockets library. There are multiple ways to do this:
$ pip install websockets
or
$ pip install uvicorn[standard]
or
$ wsproto
To test the new WebSocket route, we can use the websocat tool:
$ websocat ws://localhost:8000/echo
$ hello
Received the following text: hello
Cool! We have a working WebSocket server. Now let’s add some more functionality to our server.
To make this easier, we’ll create a simple client that will send messages to our server.
Simple Client#
Create a new file called client.py
in the same directory as app.py
.
The client will ask for your input and send it to the server.:
# This is a simple example of a WebSocket client that sends a message to the server.
# Since it's an example using the `websockets` library, and it isn't using anything
# specific to Falcon, there are no tests. Coverage is skipped for this module.
import asyncio
import websockets
async def send_message():
uri = 'ws://localhost:8000/echo'
async with websockets.connect(uri) as websocket:
while True:
message = input('Enter a message (q to exit): ')
if message.casefold() == 'q':
break
await websocket.send(message)
response = await websocket.recv()
print(response)
if __name__ == '__main__':
asyncio.run(send_message())
Run this client in a separate terminal:
$ python client.py
Enter a message: Hi
Received the following text: Hi
This will simplify testing our server.
Now let’s add some more functionality to our server.
We’ve been working with text input/output - let’s try sending sending some JSON data.
from datetime import datetime
import falcon.asgi
from falcon import WebSocketDisconnected
from falcon.asgi import Request, WebSocket
import uvicorn
app = falcon.asgi.App()
class HelloWorldResource:
async def on_get(self, req, resp):
resp.media = {'hello': 'world'}
class EchoWebSocketResource:
async def on_websocket(self, req: Request, ws: WebSocket):
try:
await ws.accept()
except WebSocketDisconnected:
return
while True:
try:
message = await ws.receive_text()
await ws.send_media({'message': message, 'date': datetime.now().isoformat()})
except WebSocketDisconnected:
return
app.add_route('/hello', HelloWorldResource())
app.add_route('/echo', EchoWebSocketResource())
if __name__ == '__main__':
uvicorn.run(app, host='localhost', port=8000)
$ python client.py
$ Enter a message: Hi
{"message": "Hi", "date": "2024-07-13T12:11:51.758923"}
Note
By default, send_media() and receive_media() will serialize to (and deserialize from) JSON for a TEXT payload, and to/from MessagePack for a BINARY payload (see also: Built-in Media Handlers).
Lets try to query for data from the server. We’ll create a new resource that will return a report based on the query.
Server side:
from datetime import datetime
import falcon.asgi
from falcon import WebSocketDisconnected
from falcon.asgi import Request, WebSocket
import uvicorn
REPORTS = {
'report1': {
'title': 'Report 1',
'content': 'This is the content of report 1',
},
'report2': {
'title': 'Report 2',
'content': 'This is the content of report 2',
},
'report3': {
'title': 'Report 3',
'content': 'This is the content of report 3',
},
'report4': {
'title': 'Report 4',
'content': 'This is the content of report 4',
},
}
app = falcon.asgi.App()
class HelloWorldResource:
async def on_get(self, req, resp):
resp.media = {'hello': 'world'}
class EchoWebSocketResource:
async def on_websocket(self, req: Request, ws: WebSocket):
try:
await ws.accept()
except WebSocketDisconnected:
return
while True:
try:
message = await ws.receive_text()
await ws.send_media({'message': message, 'date': datetime.now().isoformat()})
except WebSocketDisconnected:
return
class ReportsResource:
async def on_websocket(self, req: Request, ws: WebSocket):
try:
await ws.accept()
except WebSocketDisconnected:
return
while True:
try:
query = await ws.receive_text()
report = REPORTS.get(query, None)
print(report)
if report is None:
await ws.send_media({'error': 'report not found'})
continue
await ws.send_media({'report': report["title"]})
except WebSocketDisconnected:
return
app.add_route('/hello', HelloWorldResource())
app.add_route('/echo', EchoWebSocketResource())
app.add_route('/reports', ReportsResource())
if __name__ == '__main__':
uvicorn.run(app, host='localhost', port=8000)
We’ll also create new client app (reports_client.py), that will connect to the reports endpoint. :
import asyncio
import websockets
async def send_message():
uri = "ws://localhost:8000/reports"
async with websockets.connect(uri) as websocket:
while True:
message = input("Name of the log: ")
await websocket.send(message)
response = await websocket.recv()
print(response)
if __name__ == "__main__":
asyncio.run(send_message())
We’ve added a new resource that will return a report based on the query. The client will send a query to the server, and the server will respond with the report. If it can’t find the report, it will respond with an error message.
This is a simple example, but you can easily extend it to include more complex logic like fetching data from a database.
Middleware#
Falcon supports middleware, which can be used to add functionality to the application. For example, we can add a middleware that prints when a connection is established.
from datetime import datetime
import falcon.asgi
from falcon import WebSocketDisconnected
from falcon.asgi import Request, WebSocket
import uvicorn
REPORTS = {
'report1': {
'title': 'Report 1',
'content': 'This is the content of report 1',
},
'report2': {
'title': 'Report 2',
'content': 'This is the content of report 2',
},
'report3': {
'title': 'Report 3',
'content': 'This is the content of report 3',
},
'report4': {
'title': 'Report 4',
'content': 'This is the content of report 4',
},
}
app = falcon.asgi.App()
class LoggerMiddleware:
async def process_request_ws(self, req: Request, ws: WebSocket):
# This will be called for the HTTP request that initiates the
# WebSocket handshake before routing.
pass
async def process_resource_ws(self, req: Request, ws: WebSocket, resource, params):
# This will be called for the HTTP request that initiates the
# WebSocket handshake after routing (if a route matches the
# request).
print(f'WebSocket connection established on {req.path}')
class HelloWorldResource:
async def on_get(self, req, resp):
resp.media = {'hello': 'world'}
class EchoWebSocketResource:
async def on_websocket(self, req: Request, ws: WebSocket):
try:
await ws.accept()
except WebSocketDisconnected:
return
while True:
try:
message = await ws.receive_text()
await ws.send_media({'message': message, 'date': datetime.now().isoformat()})
except WebSocketDisconnected:
return
class ReportsResource:
async def on_websocket(self, req: Request, ws: WebSocket):
try:
await ws.accept()
except WebSocketDisconnected:
return
while True:
try:
query = await ws.receive_text()
report = REPORTS.get(query, None)
print(report)
if report is None:
await ws.send_media({'error': 'report not found'})
continue
await ws.send_media({'report': report["title"]})
except WebSocketDisconnected:
return
app.add_route('/hello', HelloWorldResource())
app.add_route('/echo', EchoWebSocketResource())
app.add_route('/reports', ReportsResource())
app.add_middleware(LoggerMiddleware())
if __name__ == '__main__':
uvicorn.run(app, host='localhost', port=8000)
Now, when you run the server, you should see a message in the console when a WebSocket connection is established.
Authentication#
Adding authentication can be done with the help of middleware as well. Authentication can be done a few ways. In this example we’ll use the First message method, as described on the websockets documentation.
There are some considerations to take into account when implementing authentication in a WebSocket server.
Updated server code:
from datetime import datetime
import logging
import pathlib
import uvicorn
from falcon import WebSocketDisconnected
import falcon.asgi
from falcon.asgi import Request
from falcon.asgi import WebSocket
logger = logging.getLogger('ws-logger')
logger.setLevel('INFO')
logger.addHandler(logging.StreamHandler())
REPORTS = {
'report1': {
'title': 'Report 1',
'content': 'This is the content of report 1',
},
'report2': {
'title': 'Report 2',
'content': 'This is the content of report 2',
},
'report3': {
'title': 'Report 3',
'content': 'This is the content of report 3',
},
'report4': {
'title': 'Report 4',
'content': 'This is the content of report 4',
},
}
app = falcon.asgi.App()
class LoggerMiddleware:
async def process_request_ws(self, req: Request, ws: WebSocket):
# This will be called for the HTTP request that initiates the
# WebSocket handshake before routing.
pass
async def process_resource_ws(self, req: Request, ws: WebSocket, resource, params):
# This will be called for the HTTP request that initiates the
# WebSocket handshake after routing (if a route matches the
# request).
logger.info('WebSocket connection established on %r', req.path)
class AuthMiddleware:
def __init__(self, protected_routes: list[str] | None = None):
if protected_routes is None:
protected_routes = []
self.protected_routes = protected_routes
async def process_request_ws(self, req: Request, ws: WebSocket):
# Opening a connection so we can receive the token
await ws.accept()
# Check if the route is protected
if req.path not in self.protected_routes:
return
token = await ws.receive_text()
if token != 'very secure token':
await ws.close(1008)
return
# Never log tokens in production
logger.info('Client with token %r Authenticated', token)
class HelloWorldResource:
async def on_get(self, req, resp):
resp.media = {'hello': 'world'}
class EchoWebSocketResource:
async def on_websocket(self, req: Request, ws: WebSocket):
while True:
try:
message = await ws.receive_text()
await ws.send_media(
{'message': message, 'date': datetime.now().isoformat()}
)
except WebSocketDisconnected:
return
class ReportsResource:
async def on_websocket(self, req: Request, ws: WebSocket):
while True:
try:
query = await ws.receive_text()
report = REPORTS.get(query, None)
logger.info('selected report: %s', report)
if report is None:
await ws.send_media({'error': 'report not found'})
continue
await ws.send_media({'report': report['title']})
except WebSocketDisconnected:
return
app.add_route('/hello', HelloWorldResource())
app.add_route('/echo', EchoWebSocketResource())
app.add_route('/reports', ReportsResource())
app.add_middleware(LoggerMiddleware())
app.add_middleware(AuthMiddleware(['/reports']))
# usually a web server, like Nginx or Caddy, should serve static assets, but
# for the purpose of this example we use falcon.
static_path = pathlib.Path(__file__).parent / 'static'
app.add_static_route('/', static_path, fallback_filename='index.html')
if __name__ == '__main__':
uvicorn.run(app, host='localhost', port=8000) # pragma: no cover
Updated client code for the reports client:
# This is a simple example of a WebSocket client that sends a message to the server.
# Since it's an example using the `websockets` library and it isn't using anything
# specific to Falcon, there are no tests. Coverage is skipped for this module.
import asyncio
import websockets
async def send_message():
uri = 'ws://localhost:8000/reports'
async with websockets.connect(uri) as websocket:
# Send the authentication token
await websocket.send('very secure token')
while True:
message = input('Name of the log (q to exit): ')
if message.casefold() == 'q':
break
await websocket.send(message)
response = await websocket.recv()
print(response)
if __name__ == '__main__':
asyncio.run(send_message())
Things we’ve changed:
Added a new middleware class AuthMiddleware that will check the token on the first message.
Opening a WebSocket connection is now handled by the middleware.
The client now sends a token as the first message, if required for that route.
Falcon was configured to serve a simple HTML page to use the echo WebSocket client for a browser.
If you try to query the reports endpoint now, everything works as expected on an authenticated route. But as soon as you remove/modify the token, the connection will be closed (after sending the first query - a downside of first-message authentication).
$ python reports_client.py
[...]
websockets.exceptions.ConnectionClosedError: received 1008 (policy violation); then sent 1008 (policy violation)
Note
This is a simple example of how to add authentication to a WebSocket server. In a real-world application, you would want to use a more secure method of authentication, such as JWT tokens.
What Now#
This tutorial is just the beginning. You can extend the server with more complex logic. For example, you could add a database to store/retrieve the reports, or add more routes to the server.
For more information on websockets in Falcon, check out the WebSocket API.