FastAPI Real-Time Features

What this skill does

# FastAPI Real-Time & Integration Features

This skill provides patterns for WebSockets, file uploads to S3, and email/notification services.

## WebSocket Implementation

### Connection Manager

```python
# app/websocket/manager.py
from fastapi import WebSocket
from typing import Dict, List, Set
import json
import asyncio

class ConnectionManager:
    def __init__(self):
        # room_id -> set of websockets
        self.rooms: Dict[str, Set[WebSocket]] = {}
        # user_id -> websocket
        self.user_connections: Dict[str, WebSocket] = {}

    async def connect(
        self,
        websocket: WebSocket,
        user_id: str,
        room_id: str = None
    ):
        await websocket.accept()
        self.user_connections[user_id] = websocket

        if room_id:
            if room_id not in self.rooms:
                self.rooms[room_id] = set()
            self.rooms[room_id].add(websocket)

    def disconnect(self, user_id: str, room_id: str = None):
        websocket = self.user_connections.pop(user_id, None)
        if websocket and room_id and room_id in self.rooms:
            self.rooms[room_id].discard(websocket)
            if not self.rooms[room_id]:
                del self.rooms[room_id]

    async def send_personal(self, user_id: str, message: dict):
        websocket = self.user_connections.get(user_id)
        if websocket:
            await websocket.send_json(message)

    async def broadcast_to_room(self, room_id: str, message: dict):
        if room_id in self.rooms:
            for websocket in self.rooms[room_id]:
                try:
                    await websocket.send_json(message)
                except Exception:
                    pass  # Connection closed

    async def broadcast_all(self, message: dict):
        for websocket in self.user_connections.values():
            try:
                await websocket.send_json(message)
            except Exception:
                pass

manager = ConnectionManager()
```

### WebSocket Endpoint

```python
# app/routes/websocket.py
from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Depends, Query
from app.websocket.manager import manager
from app.core.security import decode_token

router = APIRouter()

@router.websocket("/ws")
async def websocket_endpoint(
    websocket: WebSocket,
    token: str = Query(...)
):
    # Authenticate
    try:
        payload = decode_token(token)
        user_id = payload["sub"]
    except Exception:
        await websocket.close(code=4001, reason="Invalid token")
        return

    # Connect
    await manager.connect(websocket, user_id)

    try:
        while True:
            data = await websocket.receive_json()
            await handle_message(user_id, data)
    except WebSocketDisconnect:
        manager.disconnect(user_id)

async def handle_message(user_id: str, data: dict):
    message_type = data.get("type")

    if message_type == "join_room":
        room_id = data.get("room_id")
        websocket = manager.user_connections.get(user_id)
        if websocket and room_id:
            if room_id not in manager.rooms:
                manager.rooms[room_id] = set()
            manager.rooms[room_id].add(websocket)
            await manager.broadcast_to_room(room_id, {
                "type": "user_joined",
                "user_id": user_id
            })

    elif message_type == "message":
        room_id = data.get("room_id")
        content = data.get("content")
        await manager.broadcast_to_room(room_id, {
            "type": "message",
            "user_id": user_id,
            "content": content
        })

    elif message_type == "typing":
        room_id = data.get("room_id")
        await manager.broadcast_to_room(room_id, {
            "type": "typing",
            "user_id": user_id
        })
```

## File Upload to S3

### S3 Service

```python
# app/services/storage.py
import boto3
from botocore.config import Config
from typing import BinaryIO, Optional
import uuid
from app.config import get_settings

settings = get_settings()

class S3Storage:
    def __init__(self):
        self.client = boto3.client(
            's3',
            aws_access_key_id=settings.aws_access_key_id,
            aws_secret_access_key=settings.aws_secret_access_key,
            region_name=settings.aws_region,
            config=Config(signature_version='s3v4')
        )
        self.bucket = settings.s3_bucket

    async def upload_file(
        self,
        file: BinaryIO,
        filename: str,
        content_type: str,
        folder: str = "uploads"
    ) -> str:
        """Upload file to S3 and return URL."""
        # Generate unique key
        ext = filename.split('.')[-1] if '.' in filename else ''
        key = f"{folder}/{uuid.uuid4()}.{ext}"

        # Upload
        self.client.upload_fileobj(
            file,
            self.bucket,
            key,
            ExtraArgs={
                'ContentType': content_type,
                'ACL': 'private'
            }
        )

        return key

    def get_presigned_url(
        self,
        key: str,
        expires_in: int = 3600
    ) -> str:
        """Generate presigned URL for download."""
        return self.client.generate_presigned_url(
            'get_object',
            Params={'Bucket': self.bucket, 'Key': key},
            ExpiresIn=expires_in
        )

    def get_presigned_upload_url(
        self,
        key: str,
        content_type: str,
        expires_in: int = 3600
    ) -> dict:
        """Generate presigned URL for direct upload."""
        return self.client.generate_presigned_post(
            self.bucket,
            key,
            Fields={'Content-Type': content_type},
            Conditions=[
                {'Content-Type': content_type},
                ['content-length-range', 1, 10485760]  # 10MB max
            ],
            ExpiresIn=expires_in
        )

    async def delete_file(self, key: str):
        """Delete file from S3."""
        self.client.delete_object(Bucket=self.bucket, Key=key)

storage = S3Storage()
```

### Upload Endpoint

```python
# app/routes/uploads.py
from fastapi import APIRouter, UploadFile, File, HTTPException, Depends
from app.services.storage import storage
from app.core.security import get_current_user

router = APIRouter(prefix="/uploads", tags=["Uploads"])

ALLOWED_TYPES = ["image/jpeg", "image/png", "image/gif", "application/pdf"]
MAX_SIZE = 10 * 1024 * 1024  # 10MB

@router.post("/")
async def upload_file(
    file: UploadFile = File(...),
    user: User = Depends(get_current_user)
):
    # Validate content type
    if file.content_type not in ALLOWED_TYPES:
        raise HTTPException(400, f"File type not allowed: {file.content_type}")

    # Validate size (read first chunk)
    contents = await file.read()
    if len(contents) > MAX_SIZE:
        raise HTTPException(400, f"File too large. Max size: {MAX_SIZE} bytes")

    # Reset file position
    await file.seek(0)

    # Upload to S3
    key = await storage.upload_file(
        file.file,
        file.filename,
        file.content_type,
        folder=f"users/{user.id}"
    )

    # Save metadata to database
    upload = await Upload(
        user_id=user.id,
        filename=file.filename,
        s3_key=key,
        content_type=file.content_type,
        size=len(contents)
    ).insert()

    return {
        "id": str(upload.id),
        "filename": file.filename,
        "url": storage.get_presigned_url(key)
    }

@router.get("/presigned-upload")
async def get_presigned_upload(
    filename: str,
    content_type: str,
    user: User = Depends(get_current_user)
):
    """Get presigned URL for direct S3 upload."""
    if content_type not in ALLOWED_TYPES:
        raise HTTPException(400, f"File type not allowed")

    ext = filename.split('.')[-1] if '.' in filename else ''
    key = f"users/{user.id}/{uuid.uuid4()}.{ext}"

    presigned = storage.get_presigned_upload_url(key, content_type)

    return {
        "u