TopolearnReal-time features such as chat, live dashboards and streaming updates need persistent two-way connections.
aiohttp has first-class WebSocket support on both sides.
Server: inside a normal GET handler, create a web.WebSocketResponse(), call await ws.prepare(request) to perform the upgrade handshake, then iterate over incoming messages:
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT: ...Send with await ws.send_str(...), send_bytes, or send_json. The loop ends when the client closes; return ws from the handler.
Client: async with session.ws_connect(url) as ws: opens a connection; then the same send_* methods and async for msg in ws (or await ws.receive()) work.
Message types to handle: TEXT, BINARY, CLOSE/CLOSED/CLOSING, and ERROR (check ws.exception()). Ping/pong is automatic; heartbeat=30 on either side sends pings and closes dead connections, which is important behind load balancers that drop idle connections.
To broadcast (chat rooms, live updates), keep a set of open WebSocketResponse objects in app state, add on connect, remove in a finally block, and send to each. One slow client should not block others, so consider per-client queues for heavy traffic.
Graceful shutdown matters: open WebSockets keep the server from stopping. Close them in an on_shutdown handler with await ws.close(code=WSCloseCode.GOING_AWAY).
Example
import asyncio
import aiohttp
from aiohttp import web
from aiohttp.test_utils import TestServer
async def echo(request):
ws = web.WebSocketResponse(heartbeat=30)
await ws.prepare(request) # upgrade the connection
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
if msg.data == "bye":
await ws.close()
else:
await ws.send_json({"echo": msg.data.upper()})
return ws
async def main():
app = web.Application()
app.router.add_get("/ws", echo)
async with TestServer(app) as server, aiohttp.ClientSession() as session:
async with session.ws_connect(server.make_url("/ws")) as ws:
await ws.send_str("hello")
print(await ws.receive_json()) # {'echo': 'HELLO'}
await ws.send_str("bye")
print((await ws.receive()).type == aiohttp.WSMsgType.CLOSE) # True
asyncio.run(main())Remember
- Server: WebSocketResponse, await ws.prepare(request), async for msg in ws
- Client: async with session.ws_connect(url) as ws
- Use heartbeat= to detect dead connections; close sockets on shutdown
Watch out for
- Forgetting to remove closed connections from a broadcast set, leaking memory and erroring on send.