Debugging Python Workers

Techniques and patterns for debugging Python Workers in development and production — from real-time log streaming to request replay and progressive debugging strategies.

Real-Time Log Streaming

Stream logs from production in real-time using wrangler tail: 

# Stream logs from production in real-time
wrangler tail

# Filter logs by status, method, or IP
wrangler tail --status 500
wrangler tail --method POST
wrangler tail --ip 192.168.1.1

# Pretty print with formatting
wrangler tail --format pretty

# Save logs to file for analysis
wrangler tail > debug.log

Debugging Startup Issues

When your Worker won't start (syntax errors, import issues, binding problems), wrap your module to catch and report errors: 

# debug_wrapper.py - Wrap your entire module
import sys
import traceback

try:
    # Your imports and code here
    from js import Response, env
    import fastapi

    async def on_fetch(request):
        return Response.new("Worker started successfully!")

except Exception as e:
    # This WILL show in deployment logs
    print(f"STARTUP ERROR: {e}", file=sys.stderr)
    print(f"TRACEBACK: {traceback.format_exc()}", file=sys.stderr)

    # Create a minimal handler that reports the error
    async def on_fetch(request):
        return Response.new(
            f"Worker failed to start: {str(e)}\n\n{traceback.format_exc()}",
            status=500,
            headers={"Content-Type": "text/plain"}
        )

Progressive Debugging Strategy

When something isn't working, build up from the absolute minimum to isolate the issue: 

# Stage 1: Start with the absolute minimum
async def on_fetch(request):
    return Response.new("Stage 1: Basic handler works")

# Stage 2: Add imports one by one
from js import Response

async def on_fetch(request):
    return Response.new("Stage 2: Imports work")

# Stage 3: Test each binding
from js import Response, env

async def on_fetch(request):
    try:
        kv_test = await env.KV.get("test")
        return Response.new(f"Stage 3: KV binding works: {kv_test}")
    except Exception as e:
        return Response.new(f"KV binding failed: {e}", status=500)

# Stage 4: Finally add your logic

Request Replay for Debugging

Capture failed requests and store them in KV for later analysis and replay: 

import json
import traceback
from datetime import datetime

async def on_fetch(request):
    # Clone request for logging (body can only be read once)
    request_data = {
        "url": str(request.url),
        "method": request.method,
        "headers": dict(request.headers),
        "body": await request.text() if request.body else None
    }

    try:
        # Your logic
        response = await handle_request(request_data)
        return response

    except Exception as e:
        # Log the full request for replay
        console.error("Failed request:", json.dumps(request_data))
        console.error("Error:", str(e))

        # Store in KV for later debugging
        await env.DEBUG_REQUESTS.put(
            f"error_{datetime.now().isoformat()}",
            json.dumps({
                "request": request_data,
                "error": str(e),
                "traceback": traceback.format_exc()
            })
        )

        return Response.new("Error logged", status=500)

Health Check & Status Endpoints

Add health check endpoints that verify all bindings and dependencies: 

from datetime import datetime

async def on_fetch(request):
    url = request.url

    if url.pathname == "/health":
        # Test all bindings and dependencies
        checks = {}

        try:
            await env.KV.get("health_check")
            checks["kv"] = "ok"
        except:
            checks["kv"] = "failed"

        try:
            await env.DB.prepare("SELECT 1").first()
            checks["d1"] = "ok"
        except:
            checks["d1"] = "failed"

        try:
            await env.QUEUE.send({"test": True})
            checks["queue"] = "ok"
        except:
            checks["queue"] = "failed"

        status = 200 if all(
            v == "ok" for v in checks.values()
        ) else 503

        return Response.json({
            "status": "healthy" if status == 200 else "unhealthy",
            "checks": checks,
            "timestamp": datetime.now().isoformat()
        }, status=status)

    # Regular routing
    return await handle_request(request)

Logging Patterns

Structured logging for production debugging: 

import json
from datetime import datetime

def log(level: str, message: str, **kwargs):
    """Structured logging helper"""
    entry = {
        "timestamp": datetime.now().isoformat(),
        "level": level,
        "message": message,
        **kwargs
    }
    print(json.dumps(entry))

async def on_fetch(request):
    log("info", "Request received",
        method=request.method,
        url=str(request.url),
        user_agent=request.headers.get("User-Agent")
    )

    try:
        result = await process_request(request)
        log("info", "Request processed",
            status=200,
            duration_ms=elapsed
        )
        return result
    except Exception as e:
        log("error", "Request failed",
            error=str(e),
            traceback=traceback.format_exc()
        )
        return Response.new("Internal error", status=500)

Debugging Checklist