AioDebug

2.3.0 verified Thu Apr 16 auth: no python

AioDebug is a lightweight Python library designed for monitoring and testing `asyncio` programs. It provides features like logging slow callbacks, tracking event loop lags, and dumping stack traces for unresponsive loops, intended for always-on use in production environments. The current version is 2.3.0, released on January 4, 2022. It has an infrequent release cadence.

pip install aiodebug

Common errors

error WARNING: Executing <Task pending coro=<some_function() running at ...> ...> took X.YYY seconds ↓

cause A coroutine or callback blocked the `asyncio` event loop for longer than the threshold set in `aiodebug.log_slow_callbacks.enable()`.

fix

Review the indicated coroutine (some_function) and identify the blocking operation. Consider using asyncio.to_thread() for CPU-bound tasks, ensuring I/O operations are truly asynchronous, or breaking down long synchronous operations.

error AttributeError: module 'aiodebug' has no attribute 'some_feature_name' ↓

cause Attempting to access a `aiodebug` feature directly from the `aiodebug` top-level module (e.g., `aiodebug.enable_slow_callbacks`). The library's functionalities are organized into submodules.

fix

Import the specific submodule that contains the desired feature, for example, import aiodebug.log_slow_callbacks to use aiodebug.log_slow_callbacks.enable().

Warnings

gotcha The official GitHub repository at `qntln/aiodebug` states that the project has moved to `gitlab.com/quantlane/libs/aiodebug`. Users looking for the most up-to-date source code, issues, or contributions should refer to the GitLab instance. ↓

fix Always check the project's PyPI page or GitHub README for the most current repository link.

gotcha `aiodebug` primarily identifies *blocking* calls that prevent the `asyncio` event loop from progressing. It does not report on `asyncio`-friendly tasks that simply have a long overall execution time but yield correctly. ↓

fix Understand that `aiodebug` targets event loop blocking. For overall long-running *asynchronous* tasks, traditional profiling tools or custom instrumentation might be more appropriate.

gotcha When using `aiodebug.log_slow_callbacks`, if a parent coroutine directly `await`s a blocking function, `aiodebug` might report the parent coroutine as slow, rather than pinpointing the exact blocking sub-task. This can be misleading. ↓

fix Structure your `asyncio` code to encapsulate potentially blocking operations within their own distinct `async` functions or `Task`s where possible, and avoid direct blocking calls within critical path coroutines, or use `asyncio.to_thread` for CPU-bound work.

Install

poetry add aiodebug

Imports

log_slow_callbacks
```
import aiodebug.log_slow_callbacks
```
monitor_loop_lag
```
import aiodebug.monitor_loop_lag
```
hang_inspection
```
import aiodebug.hang_inspection
```

Quickstart

This quickstart demonstrates how to use `aiodebug.log_slow_callbacks` to identify and log `asyncio` callbacks that block the event loop for longer than a specified duration. The `slow_task` simulates a blocking operation (using `time.sleep`) which `aiodebug` will detect and report.

import asyncio
import time
import logging
import aiodebug.log_slow_callbacks

logging.basicConfig(level=logging.WARNING)

async def slow_task():
    print("Starting slow task...")
    # Simulate a blocking I/O operation or CPU-bound task
    time.sleep(0.1) 
    print("Slow task finished.")

async def main():
    print("Enabling slow callback logging...")
    aiodebug.log_slow_callbacks.enable(0.05) # Log callbacks slower than 0.05 seconds
    print("Running main task...")
    await asyncio.sleep(0.01) # Small non-blocking sleep
    await slow_task()
    print("Main task finished.")

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