WebSockets do komunikacji w czasie rzeczywistym
Twórz funkcje działające w czasie rzeczywistym w FastAPI za pomocą WebSockets: akceptuj połączenia, wymieniaj wiadomości, rozgłaszaj je do wielu klientów i poprawnie obsługuj rozłączenia, korzystając ze wzorców async.
WebSockets do komunikacji w czasie rzeczywistym to bezpłatna lekcja FastAPI Backend Development Bootcamp na CoddyKit. To lekcja 4 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej FastAPI Backend Development Bootcamp, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs FastAPI Backend Development Bootcamp zawiera 4 lekcji w sumie.
Części tej lekcji nie zostały jeszcze przetłumaczone i są wyświetlane po angielsku.
From Request/Response to Real-Time
Regular HTTP is request/response: the client asks, the server answers, the connection closes. WebSockets keep a single connection open for full-duplex, real-time messaging. Perfect for chat, live dashboards, and notifications.
Why Async Fits WebSockets
A WebSocket connection lives a long time and is mostly idle waiting for messages. FastAPI's async model lets one worker handle thousands of concurrent connections without blocking.
A Minimal WebSocket Endpoint
Declare a @app.websocket route. Accept the connection, then loop receiving and sending messages.
from fastapi import FastAPI, WebSocket
app = FastAPI()
@app.websocket('/ws')
async def ws(websocket: WebSocket):
await websocket.accept()
while True:
msg = await websocket.receive_text()
await websocket.send_text(f'echo: {msg}')Sending JSON
You can exchange structured data with receive_json and send_json instead of raw text.
data = await websocket.receive_json()
await websocket.send_json({'received': data, 'status': 'ok'})Handling Disconnects
Clients drop off. Catch WebSocketDisconnect to clean up resources instead of crashing the handler.
from fastapi import WebSocketDisconnect
try:
while True:
await websocket.receive_text()
except WebSocketDisconnect:
print('client left')A Connection Manager
To broadcast, track active connections in a manager class. Add on connect, remove on disconnect.
class Manager:
def __init__(self):
self.active = []
async def connect(self, ws):
await ws.accept()
self.active.append(ws)
def disconnect(self, ws):
self.active.remove(ws)Broadcasting to Everyone
Loop the active connections and send each one the message. This is the core of a chat room.
async def broadcast(self, message):
for connection in self.active:
await connection.send_text(message)Broadcast Logic in Plain Python
The manager is just list bookkeeping. Here is the connect/disconnect/broadcast flow simulated synchronously.
active = []
def connect(c): active.append(c)
def disconnect(c): active.remove(c)
def broadcast(msg): return [f'{c}<-{msg}' for c in active]
connect('alice'); connect('bob')
print(broadcast('hi'))
disconnect('alice')
print(broadcast('bye'))A Chat Room Endpoint
Combine the pieces: connect, broadcast each received message, and disconnect on drop.
manager = Manager()
@app.websocket('/chat')
async def chat(ws: WebSocket):
await manager.connect(ws)
try:
while True:
text = await ws.receive_text()
await manager.broadcast(text)
except WebSocketDisconnect:
manager.disconnect(ws)Scaling Beyond One Process
An in-memory manager only knows connections on its worker. To broadcast across multiple workers or servers, use a pub/sub backend like Redis to fan out messages.
Security and Limits
Protect WebSocket endpoints:
- Authenticate during the handshake (e.g. token in query or header).
- Validate and size-limit incoming messages.
- Heartbeat/ping to detect dead connections.
Quick Check
Why does an in-memory connection manager fail to broadcast correctly when you run several Uvicorn workers?
Recap
You added real-time communication:
- Accepted WebSocket connections and exchanged text/JSON.
- Handled
WebSocketDisconnectcleanly. - Built a connection manager to broadcast to many clients.
- Learned to scale with Redis pub/sub and to secure connections.
Często zadawane pytania
Czy lekcja „WebSockets do komunikacji w czasie rzeczywistym” jest bezpłatna?
Tak — pełny tekst „WebSockets do komunikacji w czasie rzeczywistym” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu FastAPI Backend Development Bootcamp, przejdź na CoddyKit PRO. Kurs FastAPI Backend Development Bootcamp zawiera 4 lekcji w sumie.
Co nauczysz się w „WebSockets do komunikacji w czasie rzeczywistym”?
Twórz funkcje działające w czasie rzeczywistym w FastAPI za pomocą WebSockets: akceptuj połączenia, wymieniaj wiadomości, rozgłaszaj je do wielu klientów i poprawnie obsługuj rozłączenia, korzystając… Ćwiczysz FastAPI Backend Development Bootcamp z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.
Czy potrzebuję doświadczenia, aby zacząć FastAPI Backend Development Bootcamp?
Nie wymagamy żadnego doświadczenia. FastAPI Backend Development Bootcamp w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 4 z 4.
Ile czasu zajmuje lekcja „WebSockets do komunikacji w czasie rzeczywistym”?
Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.
Czy mogę pisać i uruchamiać kod w tej lekcji FastAPI Backend Development Bootcamp?
Tak. Każda lekcja FastAPI Backend Development Bootcamp zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.
Wszystkie lekcje w tym kursie
- Powtórka z async/await w Pythonie
- FastAPI i operacje asynchroniczne
- Wykonywanie zadań w tle
- WebSockets do komunikacji w czasie rzeczywistym