Movatterモバイル変換

[0]ホーム
URL:

Skip to content
Join theFastAPI Cloud waiting list 🚀
Follow@fastapi onX (Twitter) to stay updated
FollowFastAPI onLinkedIn to stay updated
Subscribe to theFastAPI and friends newsletter 🎉
sponsor
sponsor
sponsor
sponsor
sponsor
sponsor
sponsor
sponsor
sponsor
sponsor
sponsor

WebSockets

You can useWebSockets withFastAPI.

Installwebsockets

Make sure you create avirtual environment, activate it, and installwebsockets (a Python library that makes it easy to use the "WebSocket" protocol):

$pipinstallwebsockets---> 100%

WebSockets client

In production

In your production system, you probably have a frontend created with a modern framework like React, Vue.js or Angular.

And to communicate using WebSockets with your backend you would probably use your frontend's utilities.

Or you might have a native mobile application that communicates with your WebSocket backend directly, in native code.

Or you might have any other way to communicate with the WebSocket endpoint.

But for this example, we'll use a very simple HTML document with some JavaScript, all inside a long string.

This, of course, is not optimal and you wouldn't use it for production.

In production you would have one of the options above.

But it's the simplest way to focus on the server-side of WebSockets and have a working example:

fromfastapiimportFastAPI,WebSocketfromfastapi.responsesimportHTMLResponseapp=FastAPI()html="""<!DOCTYPE html><html>    <head>        <title>Chat</title>    </head>    <body>        <h1>WebSocket Chat</h1>        <form action="" onsubmit="sendMessage(event)">            <input type="text" id="messageText" autocomplete="off"/>            <button>Send</button>        </form>        <ul id='messages'>        </ul>        <script>            var ws = new WebSocket("ws://localhost:8000/ws");            ws.onmessage = function(event) {                var messages = document.getElementById('messages')                var message = document.createElement('li')                var content = document.createTextNode(event.data)                message.appendChild(content)                messages.appendChild(message)            };            function sendMessage(event) {                var input = document.getElementById("messageText")                ws.send(input.value)                input.value = ''                event.preventDefault()            }        </script>    </body></html>"""@app.get("/")asyncdefget():returnHTMLResponse(html)@app.websocket("/ws")asyncdefwebsocket_endpoint(websocket:WebSocket):awaitwebsocket.accept()whileTrue:data=awaitwebsocket.receive_text()awaitwebsocket.send_text(f"Message text was:{data}")

Create awebsocket

In yourFastAPI application, create awebsocket:

fromfastapiimportFastAPI,WebSocketfromfastapi.responsesimportHTMLResponseapp=FastAPI()html="""<!DOCTYPE html><html>    <head>        <title>Chat</title>    </head>    <body>        <h1>WebSocket Chat</h1>        <form action="" onsubmit="sendMessage(event)">            <input type="text" id="messageText" autocomplete="off"/>            <button>Send</button>        </form>        <ul id='messages'>        </ul>        <script>            var ws = new WebSocket("ws://localhost:8000/ws");            ws.onmessage = function(event) {                var messages = document.getElementById('messages')                var message = document.createElement('li')                var content = document.createTextNode(event.data)                message.appendChild(content)                messages.appendChild(message)            };            function sendMessage(event) {                var input = document.getElementById("messageText")                ws.send(input.value)                input.value = ''                event.preventDefault()            }        </script>    </body></html>"""@app.get("/")asyncdefget():returnHTMLResponse(html)@app.websocket("/ws")asyncdefwebsocket_endpoint(websocket:WebSocket):awaitwebsocket.accept()whileTrue:data=awaitwebsocket.receive_text()awaitwebsocket.send_text(f"Message text was:{data}")

Technical Details

You could also usefrom starlette.websockets import WebSocket.

FastAPI provides the sameWebSocket directly just as a convenience for you, the developer. But it comes directly from Starlette.

Await for messages and send messages

In your WebSocket route you canawait for messages and send messages.

fromfastapiimportFastAPI,WebSocketfromfastapi.responsesimportHTMLResponseapp=FastAPI()html="""<!DOCTYPE html><html>    <head>        <title>Chat</title>    </head>    <body>        <h1>WebSocket Chat</h1>        <form action="" onsubmit="sendMessage(event)">            <input type="text" id="messageText" autocomplete="off"/>            <button>Send</button>        </form>        <ul id='messages'>        </ul>        <script>            var ws = new WebSocket("ws://localhost:8000/ws");            ws.onmessage = function(event) {                var messages = document.getElementById('messages')                var message = document.createElement('li')                var content = document.createTextNode(event.data)                message.appendChild(content)                messages.appendChild(message)            };            function sendMessage(event) {                var input = document.getElementById("messageText")                ws.send(input.value)                input.value = ''                event.preventDefault()            }        </script>    </body></html>"""@app.get("/")asyncdefget():returnHTMLResponse(html)@app.websocket("/ws")asyncdefwebsocket_endpoint(websocket:WebSocket):awaitwebsocket.accept()whileTrue:data=awaitwebsocket.receive_text()awaitwebsocket.send_text(f"Message text was:{data}")

You can receive and send binary, text, and JSON data.

Try it

If your file is namedmain.py, run your application with:

$fastapidevmain.py<span style="color: green;">INFO</span>:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

Open your browser athttp://127.0.0.1:8000.

You will see a simple page like:

You can type messages in the input box, and send them:

And yourFastAPI application with WebSockets will respond back:

You can send (and receive) many messages:

And all of them will use the same WebSocket connection.

UsingDepends and others

In WebSocket endpoints you can import fromfastapi and use:

  • Depends
  • Security
  • Cookie
  • Header
  • Path
  • Query

They work the same way as for other FastAPI endpoints/path operations:

fromtypingimportAnnotatedfromfastapiimport(Cookie,Depends,FastAPI,Query,WebSocket,WebSocketException,status,)fromfastapi.responsesimportHTMLResponseapp=FastAPI()html="""<!DOCTYPE html><html>    <head>        <title>Chat</title>    </head>    <body>        <h1>WebSocket Chat</h1>        <form action="" onsubmit="sendMessage(event)">            <label>Item ID: <input type="text" id="itemId" autocomplete="off" value="foo"/></label>            <label>Token: <input type="text" id="token" autocomplete="off" value="some-key-token"/></label>            <button onclick="connect(event)">Connect</button>            <hr>            <label>Message: <input type="text" id="messageText" autocomplete="off"/></label>            <button>Send</button>        </form>        <ul id='messages'>        </ul>        <script>        var ws = null;            function connect(event) {                var itemId = document.getElementById("itemId")                var token = document.getElementById("token")                ws = new WebSocket("ws://localhost:8000/items/" + itemId.value + "/ws?token=" + token.value);                ws.onmessage = function(event) {                    var messages = document.getElementById('messages')                    var message = document.createElement('li')                    var content = document.createTextNode(event.data)                    message.appendChild(content)                    messages.appendChild(message)                };                event.preventDefault()            }            function sendMessage(event) {                var input = document.getElementById("messageText")                ws.send(input.value)                input.value = ''                event.preventDefault()            }        </script>    </body></html>"""@app.get("/")asyncdefget():returnHTMLResponse(html)asyncdefget_cookie_or_token(websocket:WebSocket,session:Annotated[str|None,Cookie()]=None,token:Annotated[str|None,Query()]=None,):ifsessionisNoneandtokenisNone:raiseWebSocketException(code=status.WS_1008_POLICY_VIOLATION)returnsessionortoken@app.websocket("/items/{item_id}/ws")asyncdefwebsocket_endpoint(*,websocket:WebSocket,item_id:str,q:int|None=None,cookie_or_token:Annotated[str,Depends(get_cookie_or_token)],):awaitwebsocket.accept()whileTrue:data=awaitwebsocket.receive_text()awaitwebsocket.send_text(f"Session cookie or query token value is:{cookie_or_token}")ifqisnotNone:awaitwebsocket.send_text(f"Query parameter q is:{q}")awaitwebsocket.send_text(f"Message text was:{data}, for item ID:{item_id}")
🤓 Other versions and variants

Tip

Prefer to use theAnnotated version if possible.

fromfastapiimport(Cookie,Depends,FastAPI,Query,WebSocket,WebSocketException,status,)fromfastapi.responsesimportHTMLResponseapp=FastAPI()html="""<!DOCTYPE html><html>    <head>        <title>Chat</title>    </head>    <body>        <h1>WebSocket Chat</h1>        <form action="" onsubmit="sendMessage(event)">            <label>Item ID: <input type="text" id="itemId" autocomplete="off" value="foo"/></label>            <label>Token: <input type="text" id="token" autocomplete="off" value="some-key-token"/></label>            <button onclick="connect(event)">Connect</button>            <hr>            <label>Message: <input type="text" id="messageText" autocomplete="off"/></label>            <button>Send</button>        </form>        <ul id='messages'>        </ul>        <script>        var ws = null;            function connect(event) {                var itemId = document.getElementById("itemId")                var token = document.getElementById("token")                ws = new WebSocket("ws://localhost:8000/items/" + itemId.value + "/ws?token=" + token.value);                ws.onmessage = function(event) {                    var messages = document.getElementById('messages')                    var message = document.createElement('li')                    var content = document.createTextNode(event.data)                    message.appendChild(content)                    messages.appendChild(message)                };                event.preventDefault()            }            function sendMessage(event) {                var input = document.getElementById("messageText")                ws.send(input.value)                input.value = ''                event.preventDefault()            }        </script>    </body></html>"""@app.get("/")asyncdefget():returnHTMLResponse(html)asyncdefget_cookie_or_token(websocket:WebSocket,session:str|None=Cookie(default=None),token:str|None=Query(default=None),):ifsessionisNoneandtokenisNone:raiseWebSocketException(code=status.WS_1008_POLICY_VIOLATION)returnsessionortoken@app.websocket("/items/{item_id}/ws")asyncdefwebsocket_endpoint(websocket:WebSocket,item_id:str,q:int|None=None,cookie_or_token:str=Depends(get_cookie_or_token),):awaitwebsocket.accept()whileTrue:data=awaitwebsocket.receive_text()awaitwebsocket.send_text(f"Session cookie or query token value is:{cookie_or_token}")ifqisnotNone:awaitwebsocket.send_text(f"Query parameter q is:{q}")awaitwebsocket.send_text(f"Message text was:{data}, for item ID:{item_id}")

Info

As this is a WebSocket it doesn't really make sense to raise anHTTPException, instead we raise aWebSocketException.

You can use a closing code from thevalid codes defined in the specification.

Try the WebSockets with dependencies

If your file is namedmain.py, run your application with:

$fastapidevmain.py<span style="color: green;">INFO</span>:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

Open your browser athttp://127.0.0.1:8000.

There you can set:

  • The "Item ID", used in the path.
  • The "Token" used as a query parameter.

Tip

Notice that the querytoken will be handled by a dependency.

With that you can connect the WebSocket and then send and receive messages:

Handling disconnections and multiple clients

When a WebSocket connection is closed, theawait websocket.receive_text() will raise aWebSocketDisconnect exception, which you can then catch and handle like in this example.

fromfastapiimportFastAPI,WebSocket,WebSocketDisconnectfromfastapi.responsesimportHTMLResponseapp=FastAPI()html="""<!DOCTYPE html><html>    <head>        <title>Chat</title>    </head>    <body>        <h1>WebSocket Chat</h1>        <h2>Your ID: <span id="ws-id"></span></h2>        <form action="" onsubmit="sendMessage(event)">            <input type="text" id="messageText" autocomplete="off"/>            <button>Send</button>        </form>        <ul id='messages'>        </ul>        <script>            var client_id = Date.now()            document.querySelector("#ws-id").textContent = client_id;            var ws = new WebSocket(`ws://localhost:8000/ws/${client_id}`);            ws.onmessage = function(event) {                var messages = document.getElementById('messages')                var message = document.createElement('li')                var content = document.createTextNode(event.data)                message.appendChild(content)                messages.appendChild(message)            };            function sendMessage(event) {                var input = document.getElementById("messageText")                ws.send(input.value)                input.value = ''                event.preventDefault()            }        </script>    </body></html>"""classConnectionManager:def__init__(self):self.active_connections:list[WebSocket]=[]asyncdefconnect(self,websocket:WebSocket):awaitwebsocket.accept()self.active_connections.append(websocket)defdisconnect(self,websocket:WebSocket):self.active_connections.remove(websocket)asyncdefsend_personal_message(self,message:str,websocket:WebSocket):awaitwebsocket.send_text(message)asyncdefbroadcast(self,message:str):forconnectioninself.active_connections:awaitconnection.send_text(message)manager=ConnectionManager()@app.get("/")asyncdefget():returnHTMLResponse(html)@app.websocket("/ws/{client_id}")asyncdefwebsocket_endpoint(websocket:WebSocket,client_id:int):awaitmanager.connect(websocket)try:whileTrue:data=awaitwebsocket.receive_text()awaitmanager.send_personal_message(f"You wrote:{data}",websocket)awaitmanager.broadcast(f"Client #{client_id} says:{data}")exceptWebSocketDisconnect:manager.disconnect(websocket)awaitmanager.broadcast(f"Client #{client_id} left the chat")

To try it out:

  • Open the app with several browser tabs.
  • Write messages from them.
  • Then close one of the tabs.

That will raise theWebSocketDisconnect exception, and all the other clients will receive a message like:

Client #1596980209979 left the chat

Tip

The app above is a minimal and simple example to demonstrate how to handle and broadcast messages to several WebSocket connections.

But keep in mind that, as everything is handled in memory, in a single list, it will only work while the process is running, and will only work with a single process.

If you need something easy to integrate with FastAPI but that is more robust, supported by Redis, PostgreSQL or others, checkencode/broadcaster.

More info

To learn more about the options, check Starlette's documentation for:

[8]ページ先頭
©2009-2026 Movatter.jp