Part 4: Advanced Debugging Techniques

Introduction

Basic debugging works for synchronous code, but modern Python applications use async/await, handle exceptions across multiple layers, work with decorators, and run background tasks. I learned advanced debugging the hard way—spending hours debugging async race conditions that basic breakpoints couldn't catch.

This article covers advanced techniques: debugging async code, exception strategies, multi-threaded debugging, working with decorators and context managers, and using logpoints for production-like scenarios.

Debugging Async/Await Code

Async Python code requires different debugging approaches than synchronous code.

Basic Async Debugging

import asyncio
from aiohttp import ClientSession

async def fetch_user_data(user_id):
    async with ClientSession() as session:
        url = f"https://api.example.com/users/{user_id}"
        async with session.get(url) as response:  # ← Breakpoint here
            data = await response.json()
            return data

async def main():
    user = await fetch_user_data(123)  # ← Or breakpoint here
    print(user)

if __name__ == "__main__":
    asyncio.run(main())

Debug configuration for async code:

{
    "name": "Python: Async",
    "type": "debugpy",
    "request": "launch",
    "program": "${file}",
    "console": "integratedTerminal",
    "justMyCode": false  // See into async internals if needed
}

Debugging Concurrent Async Tasks

async def process_orders_concurrent(order_ids):
    tasks = []
    
    for order_id in order_ids:
        task = asyncio.create_task(process_order(order_id))  # ← Breakpoint
        tasks.append(task)
    
    results = await asyncio.gather(*tasks)  # ← And here
    return results

async def process_order(order_id):
    # Simulating async work
    await asyncio.sleep(0.1)  # ← Breakpoint to see which task
    order = await fetch_order(order_id)
    return order

Challenge: Multiple tasks run concurrently. Breakpoint hits multiple times.

Solution: Conditional breakpoint

# Breakpoint condition: order_id == 5
async def process_order(order_id):
    await asyncio.sleep(0.1)
    order = await fetch_order(order_id)  # Only pauses for order_id 5
    return order

Real Example: Debugging Race Conditions

I built an async service that fetched data from multiple APIs and combined results. Sometimes results were incomplete.

import asyncio
from typing import List, Dict

cache = {}

async def fetch_user(user_id: int) -> Dict:
    """Fetch user from API or cache"""
    if user_id in cache:
        print(f"Cache hit for user {user_id}")
        return cache[user_id]
    
    # Simulate API call
    await asyncio.sleep(0.1)  # ← Breakpoint here
    
    user = {"id": user_id, "name": f"User {user_id}"}
    cache[user_id] = user  # ← Race condition happens here!
    return user

async def get_users_concurrent(user_ids: List[int]):
    tasks = [fetch_user(uid) for uid in user_ids]
    return await asyncio.gather(*tasks)

async def main():
    # Request same user multiple times concurrently
    result = await get_users_concurrent([1, 2, 1, 3, 1])
    print(f"Results: {len(result)}")

Problem: When requesting user ID 1 three times concurrently, cache check fails for all three because none have finished yet.

Debugging steps:

1. Set breakpoint in fetch_user() at cache check

2. Use watch expressions:

   WATCH
     user_id
     user_id in cache
     len(cache)
     list(cache.keys())

3. Call stack shows:

   CALL STACK
     fetch_user (user_id=1)  ← Task 1
     fetch_user (user_id=1)  ← Task 2
     fetch_user (user_id=1)  ← Task 3

All three tasks check cache before any finish—race condition!

Fix with async lock:

from asyncio import Lock

cache = {}
cache_locks = {}

async def fetch_user(user_id: int) -> Dict:
    if user_id not in cache_locks:
        cache_locks[user_id] = Lock()
    
    async with cache_locks[user_id]:  # ← Only one task at a time per user
        if user_id in cache:
            return cache[user_id]
        
        await asyncio.sleep(0.1)
        user = {"id": user_id, "name": f"User {user_id}"}
        cache[user_id] = user
        return user

Debugging Event Loop Issues

# Bad: Blocking the event loop
async def slow_function():
    result = expensive_sync_operation()  # ← This blocks everything!
    return result

# Debug console during execution:
>>> asyncio.all_tasks()
{<Task pending coro=<slow_function()>>}  # Stuck here

# Good: Run sync code in executor
async def slow_function():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, expensive_sync_operation)
    return result

Exception Debugging Strategies

Enable Exception Breakpoints

VS Code can pause when exceptions are raised:

1. Run and Debug view → Breakpoints panel

2. Check "Raised Exceptions" or "Uncaught Exceptions"

def divide_numbers(a, b):
    try:
        result = a / b  # ← ZeroDivisionError pauses HERE (before except)
        return result
    except ZeroDivisionError:
        return 0

With "Raised Exceptions" enabled: Pauses at line 3 when exception occurs

With "Uncaught Exceptions" only: Doesn't pause (exception is caught)

Real Scenario: Silent Failures

def process_batch(items):
    results = []
    
    for item in items:
        try:
            result = process_item(item)  # ← Exceptions swallowed!
            results.append(result)
        except Exception as e:
            pass  # Silent failure - bad!
    
    return results

Without exception breakpoint: Failures are invisible

With exception breakpoint: Pauses on each exception, showing exactly what failed

Exception Filtering

Don't want to pause on every exception, only specific ones:

{
    "name": "Python: Current File",
    "type": "debugpy",
    "request": "launch",
    "program": "${file}",
    "console": "integratedTerminal",
    "justMyCode": true,
    "rules": [
        {
            "path": "**",
            "include": ["ValueError", "KeyError"]
        }
    ]
}

Only pauses on ValueError and KeyError, ignores others.

Debugging Exception Chains

class DataProcessingError(Exception):
    pass

def process_data(data):
    try:
        validated = validate_data(data)
    except ValueError as e:
        raise DataProcessingError("Validation failed") from e  # ← Exception chain

def validate_data(data):
    if not data.get('required_field'):
        raise ValueError("Missing required field")  # ← Original exception

Debug console at exception:

>>> e
DataProcessingError('Validation failed')

>>> e.__cause__
ValueError('Missing required field')

>>> e.__context__
ValueError('Missing required field')

# See full traceback
>>> import traceback
>>> traceback.print_exc()

Debugging with Decorators

Decorators can make debugging tricky. How do you step into decorated functions?

Basic Decorator Debugging

def log_calls(func):
    def wrapper(*args, **kwargs):
        print(f"Calling {func.__name__}")  # ← Breakpoint here
        result = func(*args, **kwargs)     # ← Or here to step into func
        print(f"Result: {result}")
        return result
    return wrapper

@log_calls
def calculate_price(amount):
    return amount * 1.1  # ← How to reach this?

Step 1: Breakpoint in wrapper at result = func(*args, **kwargs) Step 2: Press F11 (Step Into) to enter calculate_price

Debugging Decorator-Heavy Code

from functools import wraps

def require_auth(func):
    @wraps(func)
    def wrapper(user, *args, **kwargs):
        if not user.is_authenticated:  # ← Breakpoint here
            raise ValueError("Not authenticated")
        return func(user, *args, **kwargs)
    return wrapper

def rate_limit(max_calls=10):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Rate limiting logic  # ← Or here
            return func(*args, **kwargs)
        return wrapper
    return decorator

@require_auth
@rate_limit(max_calls=5)
def process_payment(user, amount):
    return charge_card(user, amount)  # ← Want to debug this

Debug strategy:

1. Breakpoint in innermost decorator wrapper (rate_limit)

2. Step Into (F11) to reach actual function

3. Use "justMyCode": false in launch.json to see into decorators

Real Example: Debugging FastAPI Dependency Injection

from fastapi import FastAPI, Depends

app = FastAPI()

def get_db():
    db = connect_database()  # ← Want to debug this
    try:
        yield db
    finally:
        db.close()

def get_current_user(db = Depends(get_db)):
    user = db.query_user()  # ← And this
    return user

@app.get("/profile")
def get_profile(user = Depends(get_current_user)):  # ← How to debug?
    return {"user": user}

Debugging FastAPI endpoints:

{
    "name": "FastAPI Debug",
    "type": "debugpy",
    "request": "launch",
    "module": "uvicorn",
    "args": [
        "main:app",
        "--reload",
        "--host", "0.0.0.0",
        "--port", "8000"
    ],
    "jinja": true,
    "justMyCode": false  // See into FastAPI internals
}

Set breakpoints in:

• get_db() to see database connection

• get_current_user() to see user loading

• get_profile() for endpoint logic

Logpoints for Production-Like Debugging

Logpoints let you output messages without stopping execution—perfect for scenarios where stopping would change behavior.

Debugging Concurrent Requests

@app.post("/order")
async def create_order(order: OrderRequest):
    # Logpoint: "Received order from user {order.user_id}"
    
    validated = validate_order(order)
    # Logpoint: "Validation result: {validated}"
    
    result = await process_order(validated)
    # Logpoint: "Processed order {result.order_id}"
    
    return result

Why logpoints instead of breakpoints?

• Server handles 100 requests/second

• Regular breakpoints would pause each request

• Logpoints show flow without stopping

Debug Console output:

Received order from user 123
Validation result: True
Processed order ORD-5001
Received order from user 456
Validation result: True
Processed order ORD-5002

Debugging Websocket Connections

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    # Logpoint: "Client connected: {websocket.client}"
    
    while True:
        data = await websocket.receive_text()
        # Logpoint: "Received: {data}"
        
        response = process_message(data)
        # Logpoint: "Sending: {response}"
        
        await websocket.send_text(response)

Regular breakpoints would disconnect clients. Logpoints show message flow without interruption.

Logpoints with Conditionals

Logpoints can include expressions:

def process_items(items):
    for i, item in enumerate(items):
        # Logpoint: "Item {i}: {item.name} - ${item.price:.2f}"
        process_item(item)

Or combined with conditional breakpoints:

def process_order(order):
    # Conditional logpoint - only logs expensive orders
    # Condition: order.total > 1000
    # Logpoint: "High-value order: {order.id} = ${order.total}"
    process_high_value(order)

Debugging Multi-Threaded Code

Python threading can be debugged in VS Code with care.

Basic Threading Debug

import threading
import time

def worker(worker_id):
    print(f"Worker {worker_id} starting")
    time.sleep(1)  # ← Breakpoint here
    print(f"Worker {worker_id} done")

threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(i,))  # ← Or here
    t.start()
    threads.append(t)

for t in threads:
    t.join()

Breakpoint behavior: Pauses ONE thread, others continue

Call stack shows all threads:

CALL STACK
  ▼ Thread 1 (paused)
      worker (line 5)
  ▼ Thread 2 (running)
  ▼ Thread 3 (running)

Debugging Thread-Safe Code

from threading import Lock

class Counter:
    def __init__(self):
        self.value = 0
        self.lock = Lock()
    
    def increment(self):
        with self.lock:  # ← Breakpoint here
            current = self.value
            # Simulate slow operation
            time.sleep(0.001)
            self.value = current + 1  # ← Watch for race conditions

Watch expressions:

WATCH
  self.value
  current
  threading.current_thread().name
  self.lock.locked()

Debugging Context Managers

class DatabaseConnection:
    def __init__(self, url):
        self.url = url
        self.connection = None
    
    def __enter__(self):
        print("Opening connection")
        self.connection = self.connect()  # ← Breakpoint here
        return self.connection
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        print("Closing connection")  # ← And here
        if self.connection:
            self.connection.close()
        return False  # Don't suppress exceptions

with DatabaseConnection("postgresql://...") as db:  # ← Step Into (F11)
    db.execute_query("SELECT * FROM users")

Step Into at with statement enters __enter__

Step Out after with block enters __exit__

Advanced Techniques

Debugging Generators

def fibonacci(n):
    a, b = 0, 1
    for _ in range(n):
        yield a  # ← Breakpoint here pauses on each yield
        a, b = b, a + b

for num in fibonacci(10):  # ← Or here
    print(num)

Each iteration hits the breakpoint. Use conditional or hit count to debug specific iterations.

Debugging Comprehensions

List comprehensions run in single expression—hard to debug:

# Hard to debug
results = [complex_transform(item) for item in items if validate(item)]

# Easier to debug - expand it
results = []
for item in items:  # ← Breakpoint
    if validate(item):  # ← See validation
        result = complex_transform(item)  # ← See transform
        results.append(result)

Post-Mortem Debugging

When code crashes, inspect the state at crash time:

import pdb
import sys

def buggy_function(data):
    return data['missing_key']  # KeyError!

try:
    buggy_function({})
except:
    import pdb
    pdb.post_mortem(sys.exc_info()[2])  # Opens debugger at exception

Or use VS Code's exception breakpoints to pause automatically.

Real-World Example: Debugging Async API Client

import asyncio
import aiohttp
from typing import List, Dict

class APIClient:
    def __init__(self, base_url, max_concurrent=5):
        self.base_url = base_url
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.session = None
    
    async def __aenter__(self):
        self.session = aiohttp.ClientSession()  # ← Breakpoint
        return self
    
    async def __aexit__(self, *args):
        await self.session.close()
    
    async def fetch(self, endpoint):
        async with self.semaphore:  # ← Limit concurrency
            url = f"{self.base_url}/{endpoint}"
            # Logpoint: "Fetching {url}"
            async with self.session.get(url) as response:
                return await response.json()
    
    async def fetch_all(self, endpoints: List[str]) -> List[Dict]:
        tasks = [self.fetch(ep) for ep in endpoints]  # ← Breakpoint
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Check for exceptions
        errors = [r for r in results if isinstance(r, Exception)]
        # Conditional breakpoint: len(errors) > 0
        
        return [r for r in results if not isinstance(r, Exception)]

async def main():
    endpoints = [f"users/{i}" for i in range(1, 20)]
    
    async with APIClient("https://api.example.com") as client:
        results = await client.fetch_all(endpoints)
        print(f"Fetched {len(results)} results")

asyncio.run(main())

Debugging strategy:

1. Logpoint in fetch(): See each URL being requested

2. Breakpoint in fetch_all(): Inspect task creation

3. Conditional breakpoint when errors occur: len(errors) > 0

4. Exception breakpoint: Catch any unhandled exceptions

5. Watch expressions:

   WATCH
     len(tasks)
     len(errors)
     [type(r).__name__ for r in results]

Best Practices

1. Use Logpoints for High-Frequency Code

# Bad: Regular breakpoint pauses 10,000 times
for item in 10000_items:  # ← Breakpoint = terrible
    process(item)

# Good: Logpoint shows progress without stopping
for item in 10000_items:  # ← Logpoint: "Processing {item.id}"
    process(item)

2. Conditional Breakpoints for Specific Cases

# Bad: Pause every iteration
for user in users:  # ← Pauses 1000 times
    if user.role == 'admin':
        configure_admin(user)

# Good: Conditional breakpoint: user.role == 'admin'
for user in users:  # ← Pauses only for admins
    configure_admin(user)

3. Exception Breakpoints for Silent Failures

Enable "Raised Exceptions" to catch swallowed exceptions in try/except blocks.

4. justMyCode Setting

{
    "justMyCode": true   // Skip library code (default)
    "justMyCode": false  // See into libraries (debugging)
}

What's Next

You now know:

• Debugging async/await and concurrent code

• Exception breakpoints and strategies

• Working with decorators and context managers

• Logpoints for production-like scenarios

• Multi-threaded debugging basics

In Part 5: Debug Configurations and Production Debugging, you'll learn creating comprehensive launch.json configurations, debugging tests, remote debugging, and investigating production issues.

---

Based on debugging complex async microservices, multi-threaded applications, and production Python systems in VS Code.