pyleak

Pyleak provides detailed stack traces to help locate the source of leaks or blocking:

• Asyncio Tasks: no_task_leaks provides the current stack trace of the leaked task. You can also include the creation stack trace by setting enable_creation_tracking=True.

  • Warning: enable_creation_tracking monkey patches asyncio.create_task and is not recommended for production.
  • The TaskLeakError exception has a leaked_tasks attribute containing LeakedTask objects.

• Event Loop Blocks: no_event_loop_blocking provides the stack trace of the code that blocked the loop, along with the duration and timestamp.

# Example: Getting creation stack trace for tasks
async with no_task_leaks(action="raise", enable_creation_tracking=True):
    await leaky_function()

# Example: Catching EventLoopBlockError
try:
    async with no_event_loop_blocking(action="raise"):
        await some_blocking_code()
except EventLoopBlockError as e:
    print(e)

The pyleak pytest plugin allows you to automatically detect leaked asyncio tasks, threads, and event loop blocking by applying the @pytest.mark.no_leaks marker to your test functions.

By default, if you use @pytest.mark.no_leaks without arguments, it monitors for all types of leaks (tasks, threads, and blocking). You can also specify exactly what to monitor using positional arguments or keyword arguments.

Supported monitoring modes:

• tasks: Detects leaked asyncio tasks.
• threads: Detects leaked threads.
• blocking: Detects event loop blocking.
• all: Enables all three detectors (default behavior when no arguments are provided).

import pytest

# Monitor everything (tasks, threads, and blocking)
@pytest.mark.no_leaks
def test_everything():
    pass

# Monitor only asyncio tasks
@pytest.mark.no_leaks("tasks")
def test_only_tasks():
    pass

# Monitor tasks and threads using keyword arguments
@pytest.mark.no_leaks(tasks=True, threads=True)
def test_specific_leaks():
    pass

Use the no_task_leaks context manager or decorator to ensure no asyncio tasks are left running after a specific block of code or function completes. This is useful for identifying tasks that were started but never awaited or cancelled.

Usage Modes:

• Async Context Manager: Wrap an async block with async with no_task_leaks():.
• Decorator: Apply @no_task_leaks to an async def function.

Configuration Options:

• action: What to do when a leak is found. Options include LeakAction.WARN, LeakAction.LOG, LeakAction.CANCEL, or LeakAction.RAISE (or their string equivalents).
• name_filter: A string or regex pattern to filter tasks by name.
• logger: A logging.Logger instance to use for reporting.
• enable_creation_tracking: If True, enables automatic tracking of task creation stacks by enabling asyncio debug mode.

# As a context manager
async with no_task_leaks():
    await some_async_function()

# As a decorator
@no_task_leaks
async def my_function():
    await some_async_function()

When no_task_leaks(action="raise") detects leaked tasks, it raises a TaskLeakError. You can inspect this error to find exactly which tasks leaked by accessing:

• e.task_count: The number of leaked tasks.
• e.leaked_tasks: A list of task information objects.

For each leaked task, you can use:

• task_info.name: The name of the task.
• task_info.format_current_stack(): A string representation of the stack where the task is currently executing.
• task_info.format_creation_stack(): A string representation of the stack where the task was originally created.
• task_info.task_ref: A reference to the actual asyncio.Task object, allowing you to call .cancel() on it.

You can also use enable_creation_tracking=True and name_filter (a compiled regex) in no_task_leaks to narrow down the search to specific types of tasks.

import asyncio
import random
import re
from pyleak import TaskLeakError, no_task_leaks

async def debug_task_leaks():
    """Example showing how to debug complex task leaks."""

    async def worker(worker_id: int, sleep_time: int):
        print(f"Worker {worker_id} starting")
        await asyncio.sleep(sleep_time)  # Simulate work
        print(f"Worker {worker_id} done")

    async def spawn_workers():
        for i in range(3):
            asyncio.create_task(worker(i, random.randint(1, 10)), name=f"worker-{i}")

    try:
        async with no_task_leaks(
            action="raise",
            enable_creation_tracking=True,
            name_filter=re.compile(r"worker-\d+"),  # Only catch worker tasks
        ):
            await spawn_workers()
            await asyncio.sleep(0.1)  # Let workers start

    except TaskLeakError as e:
        print(f"\nFound {e.task_count} leaked worker tasks:")
        for task_info in e.leaked_tasks:
            print(f"\n--- {task_info.name} ---")
            print("Currently executing:")
            print(task_info.format_current_stack())
            print("Created at:")
            print(task_info.format_creation_stack())

            # Cancel the leaked task
            if task_info.task_ref:
                task_info.task_ref.cancel()


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