Add FastAPI real-time chat template

fastapi-realtime-chat/.gitignore

@@ -0,0 +1,5 @@
+.venv/
+__pycache__/
+*.pyc
+chat.db
+.pytest_cache/

fastapi-realtime-chat/README.md

@@ -0,0 +1,49 @@
+# FastAPI Real-Time Chat
+
+This template is a real-time chat application built with FastAPI, WebSockets, simple token-based authentication, and SQLite message history. It is designed to run as a single Codesphere workspace with no external services.
+
+## Demo Users
+
+The app ships with three demo users:
+
+| Username | Password |
+| --- | --- |
+| ada | `codesphere` |
+| grace | `codesphere` |
+| linus | `codesphere` |
+
+## Run Locally
+
+```bash
+python3 -m venv .venv
+source .venv/bin/activate
+pip install -r requirements.txt
+python scripts/init_db.py
+uvicorn app.main:app --host 0.0.0.0 --port 3000 --reload
+```
+
+Open `http://localhost:3000`, sign in as a demo user, choose a room, and send messages from multiple browser tabs.
+
+## Codesphere
+
+The included `ci.yml` installs dependencies, initializes the SQLite database, runs tests, and starts the FastAPI server on port `3000`.
+
+## What This Template Demonstrates
+
+- FastAPI route handlers and static file serving.
+- WebSocket connection management.
+- Token-based authentication for HTTP and WebSocket requests.
+- SQLite persistence without extra infrastructure.
+- A browser UI that reconnects by opening a new authenticated socket per room.
+
+## Article Draft
+
+### Building a Real-Time Chat App with FastAPI WebSockets on Codesphere
+
+FastAPI is a strong choice for real-time applications because it supports both traditional HTTP endpoints and WebSockets in the same app. This template uses that combination to build a small authenticated chat application that can be deployed in one Codesphere workspace.
+
+The backend has a `/login` endpoint for demo authentication, `/rooms` and `/messages/{room}` endpoints for loading the interface, and a `/ws/{room}` WebSocket endpoint for live messages. When a user sends a message, FastAPI stores it in SQLite and broadcasts the saved payload to everyone connected to the same room.
+
+SQLite keeps the template simple: there is no database server to configure, but messages still survive process restarts. The `scripts/init_db.py` script creates the schema, and the CI pipeline runs it before starting the app.
+
+In Codesphere, the workflow is straightforward. The prepare step installs Python dependencies, initializes the database, and runs tests. The run step starts Uvicorn on port `3000`. From there, developers can replace the demo user map with a real user table, add private rooms, or connect the app to a production database.

fastapi-realtime-chat/app/__init__.py

@@ -0,0 +1 @@
+

fastapi-realtime-chat/app/auth.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+import secrets
+from typing import Annotated
+
+from fastapi import Depends, HTTPException, Query, status
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+
+DEMO_USERS = {
+    "ada": "codesphere",
+    "grace": "codesphere",
+    "linus": "codesphere",
+}
+
+_active_tokens: dict[str, str] = {}
+_bearer = HTTPBearer(auto_error=False)
+
+
+def login(username: str, password: str) -> str:
+    if DEMO_USERS.get(username) != password:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid username or password",
+        )
+
+    token = secrets.token_urlsafe(24)
+    _active_tokens[token] = username
+    return token
+
+
+def username_for_token(token: str | None) -> str | None:
+    if not token:
+        return None
+    return _active_tokens.get(token)
+
+
+def require_user(
+    credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(_bearer)],
+) -> str:
+    username = username_for_token(credentials.credentials if credentials else None)
+    if not username:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Missing or invalid token",
+        )
+    return username
+
+
+def require_socket_user(token: Annotated[str | None, Query()] = None) -> str:
+    username = username_for_token(token)
+    if not username:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Missing or invalid token",
+        )
+    return username

fastapi-realtime-chat/app/database.py

@@ -0,0 +1,57 @@
+from __future__ import annotations
+
+import sqlite3
+from pathlib import Path
+
+DATABASE_PATH = Path(__file__).resolve().parents[1] / "chat.db"
+
+
+def get_connection() -> sqlite3.Connection:
+    connection = sqlite3.connect(DATABASE_PATH)
+    connection.row_factory = sqlite3.Row
+    return connection
+
+
+def init_db() -> None:
+    with get_connection() as connection:
+        connection.execute(
+            """
+            CREATE TABLE IF NOT EXISTS messages (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                room TEXT NOT NULL,
+                username TEXT NOT NULL,
+                body TEXT NOT NULL,
+                created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+            )
+            """
+        )
+
+
+def save_message(room: str, username: str, body: str) -> dict:
+    with get_connection() as connection:
+        cursor = connection.execute(
+            """
+            INSERT INTO messages (room, username, body)
+            VALUES (?, ?, ?)
+            RETURNING id, room, username, body, created_at
+            """,
+            (room, username, body),
+        )
+        row = cursor.fetchone()
+        connection.commit()
+    return dict(row)
+
+
+def list_messages(room: str, limit: int = 50) -> list[dict]:
+    with get_connection() as connection:
+        rows = connection.execute(
+            """
+            SELECT id, room, username, body, created_at
+            FROM messages
+            WHERE room = ?
+            ORDER BY id DESC
+            LIMIT ?
+            """,
+            (room, limit),
+        ).fetchall()
+    return [dict(row) for row in reversed(rows)]

fastapi-realtime-chat/app/main.py

@@ -0,0 +1,130 @@
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+
+from fastapi import Depends, FastAPI, WebSocket, WebSocketDisconnect
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import FileResponse
+from fastapi.staticfiles import StaticFiles
+from pydantic import BaseModel, Field
+
+from app import auth, database
+
+ROOMS = ["general", "product", "support"]
+
+
+@asynccontextmanager
+async def lifespan(_: FastAPI):
+    database.init_db()
+    yield
+
+
+app = FastAPI(title="FastAPI Real-Time Chat", lifespan=lifespan)
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+app.mount("/static", StaticFiles(directory="app/static"), name="static")
+
+
+class LoginRequest(BaseModel):
+    username: str = Field(min_length=1, max_length=40)
+    password: str = Field(min_length=1, max_length=80)
+
+
+class LoginResponse(BaseModel):
+    token: str
+    username: str
+
+
+class ChatMessage(BaseModel):
+    id: int
+    room: str
+    username: str
+    body: str
+    created_at: str
+
+
+class ConnectionManager:
+    def __init__(self) -> None:
+        self.active_connections: dict[str, set[WebSocket]] = {}
+
+    async def connect(self, room: str, websocket: WebSocket) -> None:
+        await websocket.accept()
+        self.active_connections.setdefault(room, set()).add(websocket)
+
+    def disconnect(self, room: str, websocket: WebSocket) -> None:
+        connections = self.active_connections.get(room)
+        if not connections:
+            return
+        connections.discard(websocket)
+        if not connections:
+            self.active_connections.pop(room, None)
+
+    async def broadcast(self, room: str, message: dict) -> None:
+        for connection in list(self.active_connections.get(room, set())):
+            await connection.send_json(message)
+
+
+manager = ConnectionManager()
+
+
+@app.get("/")
+async def index() -> FileResponse:
+    return FileResponse("app/static/index.html")
+
+
+@app.get("/health")
+async def health() -> dict[str, str]:
+    return {"status": "ok"}
+
+
+@app.post("/login", response_model=LoginResponse)
+async def login(payload: LoginRequest) -> LoginResponse:
+    token = auth.login(payload.username, payload.password)
+    return LoginResponse(token=token, username=payload.username)
+
+
+@app.get("/rooms")
+async def rooms(username: str = Depends(auth.require_user)) -> dict:
+    return {"username": username, "rooms": ROOMS}
+
+
+@app.get("/messages/{room}", response_model=list[ChatMessage])
+async def messages(room: str, username: str = Depends(auth.require_user)) -> list[dict]:
+    _ = username
+    return database.list_messages(room)
+
+
+@app.websocket("/ws/{room}")
+async def websocket_endpoint(websocket: WebSocket, room: str) -> None:
+    token = websocket.query_params.get("token")
+    username = auth.username_for_token(token)
+    if not username:
+        await websocket.close(code=1008)
+        return
+
+    await manager.connect(room, websocket)
+    try:
+        await manager.broadcast(
+            room,
+            {
+                "id": 0,
+                "room": room,
+                "username": "system",
+                "body": f"{username} joined {room}",
+                "created_at": "",
+            },
+        )
+
+        while True:
+            payload = await websocket.receive_json()
+            body = str(payload.get("body", "")).strip()
+            if not body:
+                continue
+            saved = database.save_message(room, username, body[:500])
+            await manager.broadcast(room, saved)
+    except WebSocketDisconnect:
+        manager.disconnect(room, websocket)