0
# PyLeak
1

2
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.
3

4
## Package Information
5

6
- **Package Name**: pyleak
7
- **Language**: Python
8
- **Installation**: `pip install pyleak`
9
- **Minimum Python**: 3.9+
10

11
## Core Imports
12

13
```python
14
from pyleak import no_task_leaks, no_thread_leaks, no_event_loop_blocking
15
```
16

17
Import specific exceptions for error handling:
18

19
```python
20
from pyleak import TaskLeakError, ThreadLeakError, EventLoopBlockError, PyleakExceptionGroup
21
```
22

23
Import constants and configuration:
24

25
```python
26
from pyleak import DEFAULT_THREAD_NAME_FILTER
27
```
28

29
## Basic Usage
30

31
```python
32
import asyncio
33
import threading
34
import time
35
from pyleak import no_task_leaks, no_thread_leaks, no_event_loop_blocking
36

37
# Detect leaked asyncio tasks
38
async def detect_task_leaks():
39
    async with no_task_leaks():
40
        asyncio.create_task(asyncio.sleep(10))  # This will be detected
41
        await asyncio.sleep(0.1)
42

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

48
# Detect event loop blocking
49
async def detect_blocking():
50
    with no_event_loop_blocking():
51
        time.sleep(0.5)  # This will be detected
52
```
53

54
## Architecture
55

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

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

62
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.
63

64
## Capabilities
65

66
### AsyncIO Task Leak Detection
67

68
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.
69

70
```python { .api }
71
def no_task_leaks(
72
    action: Union[LeakAction, str] = LeakAction.WARN,
73
    name_filter: Optional[Union[str, re.Pattern]] = None,
74
    logger: Optional[logging.Logger] = None,
75
    *,
76
    enable_creation_tracking: bool = False,
77
):
78
    """Context manager/decorator that detects task leaks within its scope."""
79
```
80

81
```python { .api }
82
class TaskLeakError(LeakError):
83
    """Raised when task leaks are detected and action is set to RAISE."""
84
    def __init__(self, message: str, leaked_tasks: List[LeakedTask])
85
    
86
    leaked_tasks: List[LeakedTask]
87
    task_count: int
88
    
89
    def get_stack_summary(self) -> str: ...
90
```
91

92
[AsyncIO Task Detection](./task-detection.md)
93

94
### Thread Leak Detection
95

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

98
```python { .api }
99
def no_thread_leaks(
100
    action: str = "warn",
101
    name_filter: Optional[Union[str, re.Pattern]] = DEFAULT_THREAD_NAME_FILTER,
102
    logger: Optional[logging.Logger] = None,
103
    exclude_daemon: bool = True,
104
    grace_period: float = 0.1,
105
):
106
    """Context manager/decorator that detects thread leaks within its scope."""
107
```
108

109
```python { .api }
110
class ThreadLeakError(LeakError):
111
    """Raised when thread leaks are detected and action is set to RAISE."""
112
```
113

114
[Thread Leak Detection](./thread-detection.md)
115

116
### Event Loop Blocking Detection
117

118
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.
119

120
```python { .api }
121
def no_event_loop_blocking(
122
    action: LeakAction = LeakAction.WARN,
123
    logger: Optional[logging.Logger] = None,
124
    *,
125
    threshold: float = 0.2,
126
    check_interval: float = 0.05,
127
    caller_context: CallerContext | None = None,
128
):
129
    """Context manager/decorator that detects event loop blocking within its scope."""
130
```
131

132
```python { .api }
133
class EventLoopBlockError(LeakError):
134
    """Raised when event loop blocking is detected and action is set to RAISE."""
135
    def __init__(self, message: str, blocking_events: list[EventLoopBlock])
136
    
137
    blocking_events: list[EventLoopBlock]
138
    block_count: int
139
    
140
    def get_block_summary(self) -> str: ...
141
```
142

143
[Event Loop Blocking Detection](./blocking-detection.md)
144

145
### PyTest Integration
146

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

149
```python { .api }
150
@pytest.mark.no_leaks  # Detect all leak types
151
@pytest.mark.no_leaks(tasks=True, threads=False, blocking=True)  # Selective detection
152
def test_function(): ...
153
```
154

155
[PyTest Plugin](./pytest-integration.md)
156

157
## Shared Types and Constants
158

159
```python { .api }
160
class LeakAction(str, Enum):
161
    """Actions to take when leaks are detected."""
162
    WARN = "warn"
163
    LOG = "log"
164
    CANCEL = "cancel"
165
    RAISE = "raise"
166
```
167

168
```python { .api }
169
class PyleakExceptionGroup(ExceptionGroup, LeakError):
170
    """Combined exception for multiple leak errors."""
171
    def __init__(self, message: str, leak_errors: List[LeakError])
172
```
173

174
```python { .api }
175
DEFAULT_THREAD_NAME_FILTER: re.Pattern
176
# Compiled regex pattern that excludes asyncio threads: r"^(?!asyncio_\d+$).*"
177
```
178

179
## Error Handling
180

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

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