or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

blocking-detection.mdindex.mdpytest-integration.mdtask-detection.mdthread-detection.md
tile.json

tessl/pypi-pyleak

Detect leaked asyncio tasks, threads, and event loop blocking in Python, inspired by Go's goleak package

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/pyleak@0.1.x

To install, run

npx @tessl/cli install tessl/pypi-pyleak@0.1.0

index.mddocs/

PyLeak

A comprehensive Python library for detecting leaked asyncio tasks, threads, and event loop blocking operations. PyLeak provides context managers and pytest integration to identify resource management issues during testing and development, helping developers write more robust asynchronous code by catching resource leaks before they reach production.

Package Information

  • Package Name: pyleak
  • Language: Python
  • Installation: pip install pyleak
  • Minimum Python: 3.9+

Core Imports

from pyleak import no_task_leaks, no_thread_leaks, no_event_loop_blocking

Import specific exceptions for error handling:

from pyleak import TaskLeakError, ThreadLeakError, EventLoopBlockError, PyleakExceptionGroup

Import constants and configuration:

from pyleak import DEFAULT_THREAD_NAME_FILTER

Basic Usage

import asyncio
import threading
import time
from pyleak import no_task_leaks, no_thread_leaks, no_event_loop_blocking

# Detect leaked asyncio tasks
async def detect_task_leaks():
    async with no_task_leaks():
        asyncio.create_task(asyncio.sleep(10))  # This will be detected
        await asyncio.sleep(0.1)

# Detect leaked threads  
def detect_thread_leaks():
    with no_thread_leaks():
        threading.Thread(target=lambda: time.sleep(10)).start()  # This will be detected

# Detect event loop blocking
async def detect_blocking():
    with no_event_loop_blocking():
        time.sleep(0.5)  # This will be detected

Architecture

PyLeak is built around three core detection capabilities, each implemented as context managers that can also be used as decorators:

  • Task Leak Detection: Monitors asyncio tasks for unfinished operations that could cause memory leaks
  • Thread Leak Detection: Tracks thread lifecycle to identify threads that aren't properly cleaned up
  • Event Loop Blocking Detection: Monitors event loop responsiveness to catch synchronous operations blocking async code

All detectors support multiple action modes (warn, log, cancel, raise) and provide detailed stack trace information for debugging. The library integrates seamlessly with pytest through a plugin system for automated leak detection in test suites.

Capabilities

AsyncIO Task Leak Detection

Detects unfinished asyncio tasks that could cause memory leaks or prevent graceful shutdown, with detailed stack trace information showing where tasks are executing and where they were created.

def no_task_leaks(
    action: Union[LeakAction, str] = LeakAction.WARN,
    name_filter: Optional[Union[str, re.Pattern]] = None,
    logger: Optional[logging.Logger] = None,
    *,
    enable_creation_tracking: bool = False,
):
    """Context manager/decorator that detects task leaks within its scope."""
class TaskLeakError(LeakError):
    """Raised when task leaks are detected and action is set to RAISE."""
    def __init__(self, message: str, leaked_tasks: List[LeakedTask])
    
    leaked_tasks: List[LeakedTask]
    task_count: int
    
    def get_stack_summary(self) -> str: ...

AsyncIO Task Detection

Thread Leak Detection

Monitors thread lifecycle to identify threads that aren't properly cleaned up, helping prevent resource exhaustion and ensuring proper application termination.

def no_thread_leaks(
    action: str = "warn",
    name_filter: Optional[Union[str, re.Pattern]] = DEFAULT_THREAD_NAME_FILTER,
    logger: Optional[logging.Logger] = None,
    exclude_daemon: bool = True,
    grace_period: float = 0.1,
):
    """Context manager/decorator that detects thread leaks within its scope."""
class ThreadLeakError(LeakError):
    """Raised when thread leaks are detected and action is set to RAISE."""

Thread Leak Detection

Event Loop Blocking Detection

Monitors asyncio event loop responsiveness to detect when synchronous operations block the event loop, providing stack traces showing exactly what code is causing the blocking.

def no_event_loop_blocking(
    action: LeakAction = LeakAction.WARN,
    logger: Optional[logging.Logger] = None,
    *,
    threshold: float = 0.2,
    check_interval: float = 0.05,
    caller_context: CallerContext | None = None,
):
    """Context manager/decorator that detects event loop blocking within its scope."""
class EventLoopBlockError(LeakError):
    """Raised when event loop blocking is detected and action is set to RAISE."""
    def __init__(self, message: str, blocking_events: list[EventLoopBlock])
    
    blocking_events: list[EventLoopBlock]
    block_count: int
    
    def get_block_summary(self) -> str: ...

Event Loop Blocking Detection

PyTest Integration

Automatic leak detection in test suites using pytest markers, with configurable detection parameters and seamless integration with existing test frameworks.

@pytest.mark.no_leaks  # Detect all leak types
@pytest.mark.no_leaks(tasks=True, threads=False, blocking=True)  # Selective detection
def test_function(): ...

PyTest Plugin

Shared Types and Constants

class LeakAction(str, Enum):
    """Actions to take when leaks are detected."""
    WARN = "warn"
    LOG = "log"
    CANCEL = "cancel"
    RAISE = "raise"
class PyleakExceptionGroup(ExceptionGroup, LeakError):
    """Combined exception for multiple leak errors."""
    def __init__(self, message: str, leak_errors: List[LeakError])
DEFAULT_THREAD_NAME_FILTER: re.Pattern
# Compiled regex pattern that excludes asyncio threads: r"^(?!asyncio_\d+$).*"

Error Handling

All detectors can raise specific exception types when action="raise":

  • TaskLeakError: Contains detailed information about leaked tasks including stack traces
  • ThreadLeakError: Raised for thread leaks
  • EventLoopBlockError: Contains detailed information about blocking events including stack traces
  • PyleakExceptionGroup: Combines multiple leak errors when using combined detection