@@ -123,3 +123,38 @@ and then close the connection.
123
123
124
124
This requires the ASGI server to support the WebSocket Denial Response
125
125
extension. If it is not supported a ` RuntimeError ` will be raised.
126
+
127
+ In the context of ` Starlette ` , you can also use the ` HTTPException ` to achieve the same result.
128
+
129
+ ``` python
130
+ from starlette.applications import Starlette
131
+ from starlette.exceptions import HTTPException
132
+ from starlette.routing import WebSocketRoute
133
+ from starlette.websockets import WebSocket
134
+
135
+
136
+ def is_authorized (subprotocols : list[str ]):
137
+ if len (subprotocols) != 2 :
138
+ return False
139
+ if subprotocols[0 ] != " Authorization" :
140
+ return False
141
+ # Here we are hard coding the token, in a real application you would validate the token
142
+ # against a database or an external service.
143
+ if subprotocols[1 ] != " token" :
144
+ return False
145
+ return True
146
+
147
+
148
+ async def websocket_endpoint (websocket : WebSocket):
149
+ subprotocols = websocket.scope[" subprotocols" ]
150
+ if not is_authorized(subprotocols):
151
+ raise HTTPException(status_code = 401 , detail = " Unauthorized" )
152
+ await websocket.accept(" Authorization" )
153
+ await websocket.send_text(" Hello, world!" )
154
+ await websocket.close()
155
+
156
+
157
+ app = Starlette(debug = True , routes = [WebSocketRoute(" /ws" , websocket_endpoint)])
158
+ ```
159
+
160
+ <!-- Test the above with `npx wscat -c ws://localhost:8000/ws -s Authorization -s token` -->
0 commit comments