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

Question

I'm trying to communicate with a fingerprint device. Actaully it sends data through 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.

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, 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 answer.

Using WebSocket Protocol

app.py

from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect
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:
        print("Client disconnected")   
 
if __name__ == '__main__':
    uvicorn.run(app, host='127.0.0.1', port=8000)

test.py

import websockets
import asyncio

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