How to send and receive data in XML format using WebSockets in FastAPI?

Question

I'm trying to communicate with a fingerprint device. Actaully it sends data through a websocket connection. So, I think I can communicate with the device using webscokets. Here I'm using FastAPI, but it only accepts JSON data. The problem is that I need to handle XML data, however, I do not know how to send and accept data in XML format.

Szabo

Answer 1

FastAPI can accept and validate other types of data as well, not only JSON as you stated. Have a look at the documentation. Regarding XML, as FastAPI is actually Starlette underneath, you can use Starlette's Request object directly to read the request body as bytes (you might find this answer helpful as well), and return a custom Response with the XML data (if required). You can check if the incoming request is of the required Content-Type, and if so, let it through; otherwise, you could raise an HTTPException. Below is a working example using Python requests on client side and a normal HTTP endpoint on server side.

Using HTTP Protocol

app.py

from fastapi import FastAPI, Response, Request, HTTPException

app = FastAPI()

@app.post("/submit")
async def submit(request: Request):
    content_type = request.headers['Content-Type']
    if content_type == 'application/xml':
        body = await request.body()
        return Response(content=body, media_type="application/xml")
    else:
        raise HTTPException(status_code=400, detail=f'Content type {content_type} not supported')

test.py

import requests

body = """<?xml version='1.0' encoding='utf-8'?><a>б</a>"""
headers = {'Content-Type': 'application/xml'}
url = 'http://127.0.0.1:8000/submit'
r = requests.post(url, data=body.encode('utf-8'), headers=headers)
print(r.content)

In websockets, you can use send_bytes() and receive_bytes() for the communication, as described in Starlette's documentation, allowing you to send and receive (byte encoded) XML data as well. If you would like to perform validation on the received XML data, have a look at this.

Using WebSocket Protocol

Related answers can be found here and here, which also demonstrate how to re-establish a connection on client side, if it was terminated.

app.py

from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect
from websockets.exceptions import ConnectionClosed
import uvicorn

app = FastAPI()

@app.websocket("/ws")
async def get_stream(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            contents = await websocket.receive_bytes()
            print(str(contents, 'utf-8'))
    except (WebSocketDisconnect, ConnectionClosed):
        print("Client disconnected")   
 
if __name__ == '__main__':
    uvicorn.run(app, host='127.0.0.1', port=8000)

test.py

from websockets.exceptions import ConnectionClosed
import websockets
import asyncio

async def main():
    url = 'ws://127.0.0.1:8000/ws'
    
    async with websockets.connect(url) as ws:
        try:
            while True:
                b = bytes("<?xml version='1.0' encoding='utf-8'?><a>б</a>", 'utf-8')
                await ws.send(b)
                await asyncio.sleep(1)
        except ConnectionClosed:
            print("Server disconnected")
            
asyncio.run(main())

Finance