Question: Fast api (with websockets) and UDP

[kllr]

Hi,

We want to create an application that receives a message from IOT devices via UDP.
When a message is received this should be stored in a database and a websocket broadcast should be sent to some clients.

Is it possible to receive messages via UDP on a FastApi route and then send a message using websockets? I'm using the Postgresql fullstack example right now.

kind regards,

[nsidnev]

@kllr
Hi! Neither Starlette nor FastAPI allow direct processing of UDP messages.

But at the same time you can open asyncio UDP socket and process the traffic in it.

Here is an example that you can use to start:

import asyncio
from typing import Dict, Tuple

from fastapi import FastAPI
from starlette.endpoints import WebSocket, WebSocketEndpoint

UDP_PORT = 8001

app = FastAPI()
ws_clients: Dict[str, WebSocket] = {}

async def send_info_to_client(ws_client: WebSocket, data: bytes) -> None:
    await ws_client.send_json({"recevied": str(data), "from": ws_client.client.host})


class MyUDPProtocol(asyncio.DatagramProtocol):
    def connection_made(self, transport: asyncio.DatagramTransport) -> None:
        self.transport = transport

    def datagram_received(self, data: bytes, addr: Tuple[str, int]) -> None:
        ws_client = ws_clients[addr[0]]
        asyncio.create_task(send_info_to_client(ws_client, data))


@app.websocket_route("/ws-route")
class MyWSEndpoint(WebSocketEndpoint):
    async def on_connect(self, websocket: WebSocket) -> None:
        await websocket.accept()
        ws_clients[websocket.client.host] = websocket

    async def on_disconnect(self, websocket: WebSocket, close_code: int) -> None:
        ws_clients.pop(websocket.client.host)


@app.on_event("startup")
async def on_startup() -> None:
    loop = asyncio.get_running_loop()
    transport, protocol = await loop.create_datagram_endpoint(
        lambda: MyUDPProtocol(), local_addr=("0.0.0.0", UDP_PORT)
    )
    app.state.udp_transport = transport
    app.state.udp_protocol = protocol


@app.on_event("shutdown")
async def on_shutdown() -> None:
    app.state.udp_transport.close()

[tiangolo]

Thanks for the detailed example @nsidnev !

@kllr you could also think about separating both parts, receiving UDP in a separate process, probably based on @nsidnev 's example, then store whatever intermediate values have to be sent in Redis, and, in a FastAPI app, read from that Redis and set via WebSockets.

[kllr]

@nsidnev, Thanks for the great example!
I'll go and try to use it and seperate the udp process like @tiangolo suggested.
My setup is using redis as a broker, so that sounds like it will work.

Thanks guys and have a good weekend. I'll close the question.
Cheers!

[github-actions]

Assuming the original need was handled, this will be automatically closed now. But feel free to add more comments or create new issues or PRs.