Flask Realtime

---
name: flask-realtime
description: Conventions for a Flask REST API with WebSockets (Flask-SocketIO). Invoke when atlas shows `flask` in pyproject/requirements and any of `flask-socketio`, `eventlet`, `gevent`, `python-socketio`. Covers blueprint structure, SocketIO room patterns, JWT+Flask-Login auth, SQLAlchemy+Alembic migration discipline, Celery/RQ background jobs, pytest fixtures, and the eventlet/gevent/ASGI decision.
when_to_use: When the project uses Flask + Flask-SocketIO; covers app factory, blueprints, SocketIO rooms, JWT/Flask-Login split, Alembic discipline.
disable-model-invocation: false
---

# Flask REST + WebSockets

## When to invoke
- Atlas `frameworks.api == "Flask"` AND any of `flask-socketio`, `eventlet`, `gevent`, `python-socketio` in `pyproject.toml`/`requirements.txt`.
- User mentions Flask + "realtime", "websockets", "Socket.IO", "presence", "live updates".

## Key conventions
1. **Blueprints, not one giant `app.py`.** One blueprint per bounded context (`auth`, `users`, `orders`). Register in an app factory `create_app()`. No module-level `app = Flask(__name__)`.
2. **App factory + extension singletons.** Instantiate `db = SQLAlchemy()`, `socketio = SocketIO()`, `migrate = Migrate()`, `jwt = JWTManager()` at module top, then `.init_app(app)` inside `create_app()`. Lets tests build isolated apps.
3. **Flask-SocketIO with `socketio.run(app)` in dev, gunicorn + worker class in prod.**
   - **eventlet** (`gunicorn -k eventlet -w 1`) — mature, monkey-patches stdlib, the default for Flask-SocketIO.
   - **gevent** + `gevent-websocket` — similar; pick if you already use gevent elsewhere.
   - **ASGI via `asgiref.WsgiToAsgi`** — only when migrating to Quart/FastAPI; not ideal for SocketIO.
   - **One worker** when using sticky sessions; **N workers + Redis message queue** when scaling (`SocketIO(message_queue='redis://...')`).
4. **Auth split**: Flask-Login (server-rendered admin views, cookie session) and Flask-JWT-Extended (REST API for the SPA). Don't mix on the same route. WebSocket auth: read the JWT from the `auth` payload on `connect`, NOT the query string.
5. **SocketIO rooms = your authorization boundary.** A user joins `room=user:<id>` on connect. Broadcasts target rooms, never global. No `emit(..., broadcast=True)` without a room filter.
6. **SQLAlchemy + Alembic migrations are expand/contract.** See `.claude/skills/db-migration/SKILL.md`. Every migration has a downgrade unless data loss is documented in an ADR. Generate with `flask db migrate -m "..."` then ALWAYS hand-edit — autogen misses indexes, enums, server defaults.
7. **Background jobs**: Celery (broker + result backend, sturdy) for heavy + retried work, RQ (Redis-only, simple) for light. Don't do "background" work in the request thread — Flask-SocketIO already steals workers.
8. **Tests: pytest + `factory_boy` + Flask test client + `pytest-flask` + `pytest-socketio`.** App-factory enables `@pytest.fixture` returning an isolated app per test. Use a transactional rollback fixture for the DB.

## Canonical structure
```
src/
  myapp/
    __init__.py          # create_app() factory
    extensions.py        # db, socketio, migrate, jwt singletons
    config.py            # Config / TestConfig / ProdConfig
    blueprints/
      auth/{__init__.py, routes.py, schemas.py, services.py}
      users/{...}
      orders/{...}
    realtime/
      __init__.py        # register_socketio_handlers(socketio)
      events.py          # @socketio.on('...') handlers
      rooms.py           # room name helpers
    models/              # SQLAlchemy declarative
    tasks/               # Celery tasks
    cli.py               # flask custom commands
migrations/              # alembic
tests/
  conftest.py            # app, client, db, socketio_client fixtures
  factories.py
  test_auth.py
  test_realtime.py
pyproject.toml
gunicorn.conf.py
```

## Common pitfalls
1. **Forgetting `eventlet.monkey_patch()` BEFORE any other import.** Causes `psycopg2` / `requests` to block the event loop. Put it at the very top of `wsgi.py` (and only there).
2. **JWT in WebSocket query string.** Ends up in proxy access logs. Pass via the SocketIO `auth` dict on the client: `io({ auth: { token } })` → read in `@socketio.on('connect')`.
3. **Broadcasting without a room.** `emit('update', data, broadcast=True)` leaks across tenants. Always `room=...` or `to=...`.
4. **Autogenerated Alembic migrations shipped untouched.** Misses: enum changes, index drops, server defaults, server-side check constraints. Diff the migration, run on a prod-shaped DB, then ship.
5. **Sync DB calls inside SocketIO event handlers.** Under eventlet, SQLAlchemy must use a driver that yields (psycopg2 monkey-patched is fine; cx_Oracle is not). Long queries block the worker.
6. **N+1 in JSON serializers.** `User.query.all()` then `[u.orders for u in users]` issues N queries. Use `joinedload(User.orders)` or marshmallow with `select_related`-style nesting.

## References
- Flask app factory: https://flask.palletsprojects.com/en/latest/patterns/appfactories/
- Flask-SocketIO: https://flask-socketio.readthedocs.io
- Flask-JWT-Extended: https://flask-jwt-extended.readthedocs.io
- Alembic ops: https://alembic.sqlalchemy.org/en/latest/ops.html
- Celery + Flask: https://flask.palletsprojects.com/en/latest/patterns/celery/
- pytest-socketio: https://python-socketio.readthedocs.io/en/latest/server.html#testing

## Hello world (room-scoped event)
```python
# src/myapp/realtime/events.py
from flask import current_app
from flask_jwt_extended import decode_token
from ..extensions import socketio

@socketio.on('connect')
def on_connect(auth):
    if not auth or 'token' not in auth:
        return False                          # reject handshake
    try:
        claims = decode_token(auth['token'])
    except Exception:
        return False
    user_id = claims['sub']
    socketio.server.enter_room(request.sid, f'user:{user_id}')

@socketio.on('subscribe_order')
def on_subscribe(payload):
    order_id = payload['order_id']
    # ... authorize: does current user own this order?
    socketio.server.enter_room(request.sid, f'order:{order_id}')

# elsewhere, when an order changes:
socketio.emit('order.updated', order.to_dict(), to=f'order:{order.id}')
```