FastAPI WebSocket Notifications

Read this in Russian

FastAPI WebSocket server that broadcasts test notifications to connected clients and shuts down gracefully: each uvicorn worker waits for its own connections to close or forces shutdown after 30 minutes.

Requirements

• Python 3.11+

Setup

python3 -m venv .venv
source .venv/bin/activate
pip install -e .[dev]

Run

uvicorn app.main:app --reload
# or multi-worker
uvicorn app.main:app --workers 4

WebSocket endpoint: ws://localhost:8000/ws

Manual check

1. Connect with a client (e.g., wscat):

wscat -c ws://localhost:8000/ws

2. Every 10 seconds you should receive "test notification".
3. Disconnect (Ctrl+C in wscat) — logs will show active count decreasing.

Graceful shutdown

• Send SIGINT (Ctrl+C in the uvicorn terminal) or SIGTERM.
• Sequence:
  1. Background notification loop stops.
  2. Process waits for ConnectionManager to become idle (idle_event) or 30 minutes to elapse.
  3. If clients remain after the timeout, it sends close(1001, "Server shutdown") and exits.
• With multiple workers, each process manages only its own connections; no global sync by design.

Logging

• Console and logs/app.log.
• Rotation: 10 MB per file, up to 5 backups.

Tests

pytest

Formatting and linting

ruff check .
black .

Notes

• To change the notification interval, adjust notification_loop in app/main.py.
• Keep dependency versions pinned in pyproject.toml when upgrading.

License: MIT