Pyleak

0.2.0 · active · verified Thu Apr 16

Pyleak is a Python library inspired by Go's goleak, designed to detect leaked asyncio tasks, threads, and event loop blocking in asynchronous Python applications. The current version is 0.2.0, with frequent minor and patch releases, indicating active development.

Common errors

```
TypeError: object async_generator_wrapper can't be used in 'await' expression
```
cause Attempting to use `no_event_loop_blocking` as a decorator or regular function call after v0.2.0.

fix As of v0.2.0, `no_event_loop_blocking` must be used as an `async with` context manager. Change `@no_event_loop_blocking` or `no_event_loop_blocking()` to `async with no_event_loop_blocking():`.
```
AssertionError: Leaks detected: X asyncio tasks
```
cause Pyleak has identified uncompleted asyncio tasks, unjoined threads, or unclosed event loops when its detector context manager exited.

fix Review your application code to ensure all asyncio tasks are properly awaited or cancelled, all threads are joined, and all resources are closed before the `LeakDetector` context exits. If a task is intentionally long-lived during a test, consider allowing it explicitly, but this is generally discouraged for leak detection.
```
RuntimeError: no_event_loop_blocking must be used within an asyncio event loop.
```
cause The `no_event_loop_blocking` context manager was called from a synchronous function or outside an active asyncio event loop.

fix Ensure that code using `no_event_loop_blocking` is part of an `async` function and is executed by an active asyncio event loop, typically by calling `asyncio.run()` or `loop.run_until_complete()`.
```
Leaks are not being detected, even though I suspect they exist.
```
cause Asyncio's default debugging mode is disabled, which can limit the fidelity of stack traces and the accuracy of leak detection.

fix Enable asyncio debugging by running your application with `asyncio.run(main(), debug=True)` or by setting the environment variable `PYTHONASYNCIODEBUG=1`.

Warnings

breaking `no_event_loop_blocking` API changed from decorator/function to an async context manager.
Fix: Update usage from `@no_event_loop_blocking` or `no_event_loop_blocking()` to `async with no_event_loop_blocking():`.
gotcha For best results and detailed stack traces when detecting asyncio leaks or blocking, it is recommended to enable asyncio debug mode.
Fix: Run your application with `asyncio.run(main(), debug=True)` or set the `PYTHONASYNCIODEBUG` environment variable to `1`.
gotcha Using `LeakDetector` without properly shutting down or cancelling all expected long-running background tasks will result in false positives (AssertionError).
Fix: Ensure all asyncio tasks, threads, and resources are correctly cleaned up or awaited before the `LeakDetector` context exits. For intentional long-lived tasks, consider using `pyleak.set_allow_extra_tasks` for specific test scenarios, though this should be used cautiously.
gotcha `CombinedLeakDetector` in versions prior to 0.1.16 could lead to false positive warnings.
Fix: Upgrade pyleak to version 0.1.16 or newer to benefit from fixes addressing false positives in `CombinedLeakDetector`.

Install

pip install pyleak Install pyleak

Imports

LeakDetector
```
from pyleak import LeakDetector
```
no_event_loop_blocking
wrong:
```
@no_event_loop_blocking
async def func(): ...
```
correct:
```
from pyleak import no_event_loop_blocking
```
As of v0.2.0, no_event_loop_blocking must be used as an async context manager, not a decorator.
CombinedLeakDetector
```
from pyleak import CombinedLeakDetector
```

Quickstart

This example demonstrates how to use `LeakDetector` to identify unawaited or long-running asyncio tasks that are not properly shut down, which pyleak considers a leak. Running this code will raise an `AssertionError` at the end of the `async with LeakDetector():` block.

import asyncio
from pyleak import LeakDetector

async def create_leak():
    # A task that runs forever, simulating a leak
    async def forever_task():
        while True:
            await asyncio.sleep(100)
    asyncio.create_task(forever_task())

async def main():
    async with LeakDetector():
        await create_leak()
        print("This will run, but a leak was created.")
    # LeakDetector exits, will raise AssertionError if leaks are found
    print("LeakDetector exited, no assertion was raised (this line won't be reached if a leak is present).")

if __name__ == "__main__":
    # To see the AssertionError from the leak:
    # asyncio.run(main())
    
    # To avoid the assertion for demonstration:
    try:
        asyncio.run(main())
    except AssertionError as e:
        print(f"Caught expected leak: {e}")

view raw JSON →