Debugging Asynchronous Code: async and await

Async code introduces concurrency complexities. Common issues: tasks not awaited, race conditions, or event loop misuse.

Example: a simple async bug (forgetting to await)

# async_bug.py
import asyncio
async def slow_double(x):
    await asyncio.sleep(0.1)
    return x  2

async def main():
    # Bug: missing await causes coroutine object, not result.
    results = [slow_double(i) for i in range(3)]
    print("results:", results)  # prints coroutine objects
    # Correct approach:
    awaited = await asyncio.gather((slow_double(i) for i in range(3)))
    print("awaited:", awaited)
if __name__ == "__main__":
    asyncio.run(main())

Line-by-line:

• slow_double: async function that sleeps and returns doubled value.
• results = [slow_double(i) for i in range(3)]: creates coroutine objects; not executed.
• asyncio.gather + await: runs coroutines concurrently and awaits results.

• Use logging inside coroutines to understand scheduling.
• Use asyncio.run, not loop.run_until_complete in simple scripts.
• Use pytest-asyncio or pytest-aiohttp for testing async functions.
• For live debugging, use aiomonitor or asyncio.Task.get_stack to inspect tasks.

Profiling for Performance (CPU & Time)

When bugs are performance-related, use profilers.

Simple usage with cProfile and pstats:

# profile_example.py
import cProfile
import pstats
from io import StringIO
from sample_module import heavy_work  # imagine a CPU-heavy function
pr = cProfile.Profile()
pr.enable()
heavy_work()
pr.disable()
s = StringIO()
ps = pstats.Stats(pr, stream=s).sort_stats("cumtime")
ps.print_stats(20)
print(s.getvalue())

This shows functions consuming the most cumulative time.

Use timeit for microbenchmarks and line_profiler for per-line timings. Use sampling profilers (py-spy) for low overhead in production.

Memory Debugging: tracemalloc and memory_profiler

Find memory allocation hotspots.

Example with tracemalloc:

# tracemalloc_example.py
import tracemalloc
def allocate():
    a = [i for i in range(100000)]
    b = {i: str(i) for i in range(50000)}
    return a, b
tracemalloc.start()
snapshot1 = tracemalloc.take_snapshot()
allocate()
snapshot2 = tracemalloc.take_snapshot()
top_stats = snapshot2.compare_to(snapshot1, "lineno")
for stat in top_stats[:10]:
    print(stat)

Line-by-line:

• tracemalloc.start(): begin tracing allocations.
• take_snapshot(): capture memory state.
• compare_to: shows differences by line number.

• tracemalloc tracks Python memory allocations — C extensions may not be visible.
• For line-by-line memory usage, use memory_profiler (@profile decorator) with psutil installed.

Post-mortem Debugging and Crash Analysis

If the application crashes in production, you may want a post-mortem analysis.

• Use sys.excepthook to log unhandled exceptions with full tracebacks.
• Use faulthandler to dump Python traceback on a signal (SIGUSR1) or on fatal errors: faulthandler.dump_traceback_later or faulthandler.enable().
• Configure core dumps via OS and analyze with gdb + python extensions for low-level crashes.

import faulthandler, sys
faulthandler.enable()  # writes to stderr by default
Or redirect:
with open("faulthandler.log", "w") as f:
    faulthandler.enable(file=f)

Debugging Real-World Scripts: Automating Repetitive Tasks

Many developers write scripts to automate file processing, ETL, or web scraping. Ensure those scripts are debuggable.

Example: CSV processing automation with logging and tests

# process_csv.py
import csv
import logging
logger = logging.getLogger(__name__)
def process_row(row):
    # Example transformation that may raise KeyError
    try:
        value = int(row["quantity"]) * float(row["price"])
    except KeyError as e:
        logger.error("Missing column: %s in row: %r", e, row)
        raise
    except ValueError as e:
        logger.warning("Invalid numeric value in row: %r", row)
        return None
    return {"sku": row.get("sku"), "total": value}
def process_file(path):
    results = []
    with open(path, newline="") as csvfile:
        reader = csv.DictReader(csvfile)
        for row in reader:
            r = process_row(row)
            if r:
                results.append(r)
    return results

Debugging strategy:

• Add a small unit test with a malformed CSV to reproduce.
• Use logging to show which rows fail.
• For large files, process a small sample to iterate faster.

Debugging Interactive Dashboards with Plotly

When creating dashboards (Plotly Dash or Plotly + Flask), bugs can live in server code, client-side scripts, or network layers.

Common issues:

• Callbacks not firing (Dash) because IDs don't match.
• Race conditions between data refresh and rendering.
• Slow queries causing UI timeouts.

• Reproduce the issue with a minimal app.
• Log server-side events (incoming requests, callback triggers).
• Use browser dev tools to inspect network requests and JS errors.
• Isolate data pipeline: test the API endpoint that supplies chart data.

# plotly_api.py (simplified)
from flask import Flask, jsonify
import logging
app = Flask(__name__)
logging.basicConfig(level=logging.DEBUG)
@app.route("/data")
def data():
    logging.debug("data endpoint called")
    # Imagine heavy DB call here
    sample = {"x": [1,2,3], "y": [4,5,6]}
    return jsonify(sample)

Debugging tips:

• Test the endpoint with curl or httpie to confirm payload.
• If using async background tasks to refresh data (e.g., websockets), ensure tasks are awaited and errors are handled.

Common Pitfalls and How to Avoid Them

• Swallowing exceptions (bare except): logs vanish. Use specific exceptions, and log exceptions with stack traces.
• Mutating shared state without synchronization in concurrent code: use locks or prefer immutable data.
• Ignoring warnings: treat DeprecationWarning and ResourceWarning seriously during development.
• Not testing edge cases: include tests for invalid inputs, empties, boundary values.
• Over-optimizing before profiling: Always profile before micro-optimizing.

Advanced Tips

• Use type hints + mypy to find mismatched types before runtime.
• Enable stricter linters (flake8, pylint) to catch style and potential bugs.
• For production, use sampling profilers (py-spy) to inspect live systems with low overhead.
• Use structured logging (JSON) for easier aggregation in ELK/Graylog.
• Attach a remote debugger (debugpy for VS Code) if you need to debug code running in a container or remote host.

# remote_debug.py
import debugpy
Listen for debugger on port 5678 (do not leave this open in production)
debugpy.listen(("0.0.0.0", 5678))
print("Waiting for debugger to attach...")
debugpy.wait_for_client()
debugpy.breakpoint()
Your app logic here

Security: protect debug ports and avoid enabling this in production unless secured.

Putting It Together: Debugging Workflow Example

Scenario: An automated ETL script that feeds a Plotly dashboard sporadically stops updating. Steps:

1. Reproduce locally with sample data.
2. Add logging around ETL stages (extract, transform, load) with timestamps and IDs.
3. Run the script under a profiler to detect long-running tasks.
4. Use tracemalloc if memory spikes before failing.
5. If ETL runs asynchronously (asyncio), ensure all tasks are awaited and check for unhandled exceptions in tasks.
6. If dashboard doesn't reflect data, curl the API endpoint to verify payloads.
7. Add unit tests covering sample failure modes and integrate with CI.

Resources and References

• Official Python docs: logging, pdb, asyncio, cProfile, tracemalloc, faulthandler
• pytest documentation for testing strategies
• Plotly / Dash docs for dashboard development and debugging
• py-spy, memory_profiler, line_profiler for performance analysis
• debugpy for remote debugging

Conclusion

Effective debugging is a blend of discipline, good tooling, and iterative experiments. Use structured logging instead of print, adopt interactive debuggers for live inspection, write unit tests to capture regressions, profile only when needed, and apply async-aware techniques when working with concurrency.

Next steps:

• Pick one bug in your current project. Reproduce it, add a test, and debug using at least one tool from this guide (logging, pdb, or a profiler).
• If you automate tasks or build dashboards, instrument them with structured logs and small tests.

Further Reading

• Python logging — official docs
• pdb and the built-in breakpoint() — official docs
• asyncio — official docs (async/await patterns)
• pytest — official documentation
• Plotly Dash documentation and examples