Module 5: WebSockets and File Uploads

Project: Real-time Notifications and File Uploads

Next →

Project overview

In this project you add two fundamental capabilities to the Task Manager API you've been building since Module 1: real-time notifications via WebSocket and file uploads to attach files to tasks.

The result: any user connected to the WebSocket will instantly see when someone creates, updates, or completes a task. And every task can have attached files (PDFs, images, documents) with type and size validation.

These two features turn your API from "a CRUD that only answers when asked" into "a system that keeps everyone in sync in real time and handles multimedia content."

Project objective

By the time you complete this project:

  • ✅ Your API has a WebSocket endpoint that notifies events in real time
  • ✅ A ConnectionManager handles multiple simultaneous connections
  • ✅ Creating, updating, and completing tasks generates WebSocket notifications
  • ✅ Tasks support file uploads with type and size validation
  • ✅ Files are saved to disk with unique names
  • ✅ There's an endpoint to download attached files
  • ✅ WebSocket + background tasks work together for notifications

Technical specifications

Stack

  • Python 3.9+
  • FastAPI with uvicorn
  • python-multipart (for uploads)
  • websockets (for testing)

Updated structure

task-manager-api/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── models.py
│   ├── data.py
│   ├── dependencies.py
│   ├── websocket_manager.py       ← NEW
│   ├── routers/
│   │   ├── __init__.py
│   │   ├── tasks.py               ← Updated with WS notifications
│   │   ├── websocket.py           ← NEW
│   │   └── uploads.py             ← NEW
│   └── uploads/                   ← Directory for files
├── requirements.txt
└── README.md

Setup

pip install python-multipart
mkdir -p app/uploads

Step 1: ConnectionManager

Create app/websocket_manager.py:

from fastapi import WebSocket
from datetime import datetime


class ConnectionManager:
    def __init__(self):
        self.active_connections: list[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        if websocket in self.active_connections:
            self.active_connections.remove(websocket)

    async def send_personal(self, data: dict, websocket: WebSocket):
        await websocket.send_json(data)

    async def broadcast(self, data: dict):
        disconnected = []
        for connection in self.active_connections:
            try:
                await connection.send_json(data)
            except Exception:
                disconnected.append(connection)
        for conn in disconnected:
            self.disconnect(conn)

    async def broadcast_event(self, event_type: str, payload: dict):
        message = {
            "type": event_type,
            "data": payload,
            "timestamp": datetime.now().isoformat(),
            "connections": len(self.active_connections)
        }
        await self.broadcast(message)

    @property
    def connection_count(self) -> int:
        return len(self.active_connections)


ws_manager = ConnectionManager()

Step 2: The WebSocket router

Create app/routers/websocket.py:

from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from app.websocket_manager import ws_manager

router = APIRouter(tags=["WebSocket"])


@router.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
    await ws_manager.connect(ws)
    await ws_manager.send_personal(
        {
            "type": "connected",
            "message": "Connected to the Task Manager event stream",
            "active_connections": ws_manager.connection_count
        },
        ws
    )

    try:
        while True:
            data = await ws.receive_json()

            if data.get("action") == "ping":
                await ws_manager.send_personal({"type": "pong"}, ws)
            elif data.get("action") == "status":
                await ws_manager.send_personal(
                    {
                        "type": "status",
                        "connections": ws_manager.connection_count
                    },
                    ws
                )
    except WebSocketDisconnect:
        ws_manager.disconnect(ws)
        await ws_manager.broadcast_event("user_disconnected", {
            "connections": ws_manager.connection_count
        })

Step 3: Update the tasks router with notifications

Modify app/routers/tasks.py to add WebSocket notifications to every CRUD operation:

from fastapi import APIRouter, HTTPException, Depends
from app.models import TaskCreate, TaskUpdate, TaskPatch, TaskResponse
from app.data import tasks_db, generate_id, get_current_timestamp
from app.dependencies import get_task_or_404, pagination_params
from app.websocket_manager import ws_manager

router = APIRouter(prefix="/tasks", tags=["Tasks"])


@router.post("/", status_code=201, response_model=TaskResponse)
async def create_task(task_data: TaskCreate):
    new_task = {
        "id": generate_id(),
        "title": task_data.title,
        "description": task_data.description,
        "status": task_data.status.value if hasattr(task_data.status, 'value') else task_data.status,
        "priority": task_data.priority.value if hasattr(task_data.priority, 'value') else task_data.priority,
        "created_at": get_current_timestamp(),
        "updated_at": get_current_timestamp(),
        "attachments": []
    }
    tasks_db.append(new_task)

    await ws_manager.broadcast_event("task_created", {
        "task_id": new_task["id"],
        "title": new_task["title"],
        "priority": new_task["priority"]
    })

    return new_task


@router.get("/", response_model=list[TaskResponse])
async def list_tasks(pagination: dict = Depends(pagination_params)):
    skip = pagination["skip"]
    limit = pagination["limit"]
    return tasks_db[skip:skip + limit]


@router.get("/{task_id}", response_model=TaskResponse)
async def get_task(task: dict = Depends(get_task_or_404)):
    return task


@router.put("/{task_id}", response_model=TaskResponse)
async def update_task(task_data: TaskUpdate, task: dict = Depends(get_task_or_404)):
    task["title"] = task_data.title
    task["description"] = task_data.description
    task["status"] = task_data.status.value if hasattr(task_data.status, 'value') else task_data.status
    task["priority"] = task_data.priority.value if hasattr(task_data.priority, 'value') else task_data.priority
    task["updated_at"] = get_current_timestamp()

    await ws_manager.broadcast_event("task_updated", {
        "task_id": task["id"],
        "title": task["title"],
        "status": task["status"]
    })

    return task


@router.patch("/{task_id}", response_model=TaskResponse)
async def patch_task(task_data: TaskPatch, task: dict = Depends(get_task_or_404)):
    update_data = task_data.model_dump(exclude_unset=True)

    old_status = task.get("status")

    for field, value in update_data.items():
        if hasattr(value, 'value'):
            task[field] = value.value
        else:
            task[field] = value
    task["updated_at"] = get_current_timestamp()

    new_status = task.get("status")
    if old_status != new_status and new_status == "completed":
        await ws_manager.broadcast_event("task_completed", {
            "task_id": task["id"],
            "title": task["title"],
            "completed_at": task["updated_at"]
        })
    else:
        await ws_manager.broadcast_event("task_patched", {
            "task_id": task["id"],
            "fields_updated": list(update_data.keys())
        })

    return task


@router.delete("/{task_id}", status_code=204)
async def delete_task(task: dict = Depends(get_task_or_404)):
    tasks_db.remove(task)

    await ws_manager.broadcast_event("task_deleted", {
        "task_id": task["id"],
        "title": task["title"]
    })

Step 4: The upload router

Create app/routers/uploads.py:

import uuid
from pathlib import Path
from fastapi import APIRouter, UploadFile, File, Form, HTTPException, Depends
from fastapi.responses import FileResponse
from app.data import tasks_db
from app.dependencies import get_task_or_404
from app.websocket_manager import ws_manager

router = APIRouter(prefix="/tasks", tags=["Attachments"])

UPLOAD_DIR = Path("app/uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

ALLOWED_TYPES = {
    "image/jpeg", "image/png", "image/gif", "image/webp",
    "application/pdf",
    "text/plain", "text/csv",
    "application/msword",
    "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
}
MAX_FILE_SIZE = 5 * 1024 * 1024  # 5 MB


def generate_safe_filename(original: str) -> str:
    ext = Path(original).suffix.lower()
    return f"{uuid.uuid4().hex[:12]}{ext}"


@router.post("/{task_id}/attachments")
async def upload_attachment(
    task_id: int,
    file: UploadFile = File(...),
    description: str = Form(default=""),
    task: dict = Depends(get_task_or_404)
):
    if file.content_type not in ALLOWED_TYPES:
        raise HTTPException(400, detail={
            "error": "File type not allowed",
            "received": file.content_type,
            "allowed": list(ALLOWED_TYPES)
        })

    content = await file.read()
    if len(content) > MAX_FILE_SIZE:
        raise HTTPException(413, detail={
            "error": "File too large",
            "max_mb": MAX_FILE_SIZE / (1024 * 1024),
            "received_mb": round(len(content) / (1024 * 1024), 2)
        })

    safe_name = generate_safe_filename(file.filename)
    file_path = UPLOAD_DIR / safe_name

    with open(file_path, "wb") as f:
        f.write(content)

    attachment_info = {
        "filename": safe_name,
        "original_name": file.filename,
        "description": description,
        "content_type": file.content_type,
        "size_bytes": len(content)
    }

    if "attachments" not in task:
        task["attachments"] = []
    task["attachments"].append(attachment_info)

    await ws_manager.broadcast_event("attachment_added", {
        "task_id": task_id,
        "task_title": task["title"],
        "filename": file.filename,
        "size_kb": round(len(content) / 1024, 1)
    })

    return {
        "task_id": task_id,
        "attachment": attachment_info,
        "total_attachments": len(task["attachments"])
    }


@router.get("/{task_id}/attachments")
async def list_attachments(task: dict = Depends(get_task_or_404)):
    return {
        "task_id": task["id"],
        "attachments": task.get("attachments", []),
        "count": len(task.get("attachments", []))
    }


@router.get("/{task_id}/attachments/{filename}")
async def download_attachment(
    filename: str,
    task: dict = Depends(get_task_or_404)
):
    attachments = task.get("attachments", [])
    attachment = next(
        (a for a in attachments if a["filename"] == filename),
        None
    )

    if not attachment:
        raise HTTPException(404, detail=f"Attachment {filename} not found on this task")

    file_path = UPLOAD_DIR / filename
    if not file_path.exists():
        raise HTTPException(404, detail="File not found on disk")

    return FileResponse(
        path=file_path,
        filename=attachment["original_name"],
        media_type=attachment["content_type"]
    )

Step 5: Update main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.routers import tasks, websocket, uploads


@asynccontextmanager
async def lifespan(app):
    print("Task Manager API with WebSocket and Uploads started")
    yield
    print("Task Manager API stopped")


app = FastAPI(
    title="Task Manager API",
    description="An advanced API with real-time notifications and file uploads",
    version="2.0.0",
    lifespan=lifespan
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(tasks.router)
app.include_router(uploads.router)
app.include_router(websocket.router)


@app.get("/health")
async def health():
    from app.websocket_manager import ws_manager
    return {
        "status": "healthy",
        "websocket_connections": ws_manager.connection_count
    }

Verification

Preparation

# Terminal 1: the server
uvicorn app.main:app --reload

# Terminal 2: the WebSocket client
python -c "
import asyncio, websockets, json

async def listen():
    async with websockets.connect('ws://localhost:8000/ws') as ws:
        print('Connected to the WebSocket')
        while True:
            msg = json.loads(await ws.recv())
            print(f'[{msg[\"type\"]}] {json.dumps(msg[\"data\"], indent=2)}')

asyncio.run(listen())
"

Tests

1. Check the WebSocket connection: The client in Terminal 2 should show: [connected] ...

2. Create a task (Terminal 3):

curl -s -X POST http://localhost:8000/tasks/ \
  -H "Content-Type: application/json" \
  -d '{"title": "Test task", "description": "Testing WS", "priority": "high"}' | python -m json.tool

Terminal 2 should show: [task_created] {"task_id": ..., "title": "Test task", ...}

3. Complete the task:

curl -s -X PATCH http://localhost:8000/tasks/1 \
  -H "Content-Type: application/json" \
  -d '{"status": "completed"}' | python -m json.tool

Terminal 2 should show: [task_completed] ...

4. Upload a file:

echo "Test content" > /tmp/test.txt
curl -s -X POST http://localhost:8000/tasks/1/attachments \
  -F "file=@/tmp/test.txt" \
  -F "description=Test file" | python -m json.tool

Terminal 2 should show: [attachment_added] ...

5. List the attachments:

curl -s http://localhost:8000/tasks/1/attachments | python -m json.tool

6. Download the file:

# Use the filename from step 4
curl -s http://localhost:8000/tasks/1/attachments/{filename} -o downloaded.txt
cat downloaded.txt

7. Type validation:

# Try uploading an .exe (it should fail with 400)
echo "fake" > /tmp/test.exe
curl -s -X POST http://localhost:8000/tasks/1/attachments \
  -F "file=@/tmp/test.exe;type=application/x-msdownload" | python -m json.tool

8. Health check with the WS count:

curl -s http://localhost:8000/health | python -m json.tool

9. Delete the task:

curl -s -X DELETE http://localhost:8000/tasks/1 -w "%{http_code}"

Terminal 2 should show: [task_deleted] ...

Completeness checklist

  • ConnectionManager with connect/disconnect/broadcast
  • A working WebSocket endpoint at /ws
  • WebSocket notification on POST /tasks/
  • WebSocket notification on PUT /tasks/{id}
  • WebSocket notification on PATCH /tasks/{id}
  • A special notification when status changes to "completed"
  • WebSocket notification on DELETE /tasks/{id}
  • Upload endpoint with MIME type validation
  • Upload endpoint with size validation (5MB)
  • Files saved with unique names
  • An endpoint to list a task's attachments
  • An endpoint to download an attachment
  • WebSocket notification when a file is uploaded
  • Health check shows the WS connection count
  • Multiple WS clients receive the broadcast simultaneously

Troubleshooting

The WebSocket disconnects immediately

Check that no middleware is interfering. CORSMiddleware doesn't block WebSocket, but proxies like nginx need special configuration.

ModuleNotFoundError: No module named 'multipart'

Install it: pip install python-multipart. FastAPI requires it for UploadFile and Form().

The files aren't being saved

Check that the app/uploads/ directory exists and has write permissions. mkdir -p app/uploads.

The broadcast doesn't reach every client

Check that there's a single instance of ws_manager (imported from app.websocket_manager). If you create another instance, the clients end up in separate managers.

422 Unprocessable Entity on upload

When you use File() and Form(), the Content-Type has to be multipart/form-data, not application/json. With curl, use -F instead of -d.

file.size returns None

Read the content first: content = await file.read(), then use len(content).

The server hangs with large files

Use chunks instead of a full await file.read() for files larger than 10MB.

A WebSocket disconnection isn't detected

Make sure you're catching WebSocketDisconnect in the try/except. Without it, the error propagates and can crash things.

Resources

  1. FastAPI WebSockets — The official docs
  2. FastAPI Request Files — UploadFile
  3. FastAPI Forms and Files — Combining them
  4. Starlette WebSocket — The base API
  5. python-multipart — The parser
  6. FileResponse — Serving files

What's next?

In Module 6 you integrate EVERYTHING you learned in modules 1-5 into a complete final project: the Task Manager API with dependency injection, a modular APIRouter, advanced response models, background tasks, WebSockets, and file uploads. It's your portfolio piece that proves you've mastered advanced FastAPI.

← PreviousFile Uploads and ValidationNext →Final Project: Advanced Task Manager API