### Install pyleak

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Install the pyleak library using pip.

```bash
pip install pyleak
```

--------------------------------

### Quick Start: Detect Leaks

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Demonstrates basic usage of pyleak detectors for asyncio tasks, threads, and event loop blocking.

```python
import asyncio
from pyleak import no_task_leaks, no_thread_leaks, no_event_loop_blocking

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

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

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

--------------------------------

### Selective Detection with no_leaks Marker

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Configure specific detectors within the 'no_leaks' marker to enable or disable certain leak checks. This example enables task and blocking detection while disabling thread detection.

```python
@pytest.mark.no_leaks(tasks=True, blocking=True, threads=False)
@pytest.mark.asyncio
async def test_async_no_leaks():
    asyncio.create_task(asyncio.sleep(10))  # This will be detected
    time.sleep(0.5)  # This will be detected
    threading.Thread(target=lambda: time.sleep(10)).start()  # This will not be detected
```

--------------------------------

### Get Stack Trace for Leaked AsyncIO Tasks

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Demonstrates how to capture and print detailed stack traces for leaked asyncio tasks using `no_task_leaks` with `action='raise'`.

```python
import asyncio
from pyleak import TaskLeakError, no_task_leaks

async def leaky_function():
    async def background_task():
        print("background task started")
        await asyncio.sleep(10)

    print("creating a long running task")
    asyncio.create_task(background_task())

async def main():
    try:
        async with no_task_leaks(action="raise"):
            await leaky_function()
    except TaskLeakError as e:
        print(e)

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

--------------------------------

### Detect leaked threads with 'warn' action

Source: https://context7.com/deepankarm/pyleak/llms.txt

Use `no_thread_leaks` as a context manager to detect threads that are started but never joined. This basic usage will issue a `ResourceWarning` if leaks are found. Daemon threads are excluded by default.

```python
import threading
import time
from pyleak import ThreadLeakError, no_thread_leaks

# ── Basic context manager usage ────────────────────────────────────────────────
def example_warn():
    with no_thread_leaks(action="warn"):
        t = threading.Thread(target=lambda: time.sleep(10))
        t.start()  # never joined → ResourceWarning
        time.sleep(0.1)
```

--------------------------------

### Detect and Raise Event Loop Blocking Errors

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Use `no_event_loop_blocking` with action='raise' to catch and report event loop blocking issues. This example demonstrates how to use it within an async context manager and handle the `EventLoopBlockError`.

```python
import asyncio
import time

from pyleak import EventLoopBlockError, no_event_loop_blocking


async def some_function_with_blocking_code():
    print("starting")
    time.sleep(1)
    print("done")


async def main():
    try:
        async with no_event_loop_blocking(action="raise"):
            await some_function_with_blocking_code()
    except EventLoopBlockError as e:
        print(e)


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

--------------------------------

### no_thread_leaks

Source: https://context7.com/deepankarm/pyleak/llms.txt

Detects leaked threads that were started but never joined. It snapshots live threads before and after a block of code executes. Daemon threads are excluded by default. A grace period can be configured for threads to finish naturally.

```APIDOC
## no_thread_leaks

### Description
Snapshots the set of live threads on entry and exit of a synchronous or async block. Threads started within the block and still alive at exit are reported as leaks. Daemon threads are excluded by default.

### Usage
Can be used as a context manager.

### Parameters
- **action** (str): The action to take upon detecting a leak. Options include "warn" or "raise". The "cancel" action is not applicable to threads.
- **grace_period** (float, optional): A short window (default 0.1 seconds) for threads to finish naturally before the check runs.

### Actions
- **warn**: Issues a `ResourceWarning`.
- **raise**: Raises a `ThreadLeakError`.

### Examples

#### Basic context manager usage
```python
import threading
import time
from pyleak import no_thread_leaks

def example_warn():
    with no_thread_leaks(action="warn"):
        t = threading.Thread(target=lambda: time.sleep(10))
        t.start()  # never joined → ResourceWarning
        time.sleep(0.1)
```

#### Raise exception on leak
```python
import threading
import time
from pyleak import ThreadLeakError, no_thread_leaks

def example_raise():
    try:
        with no_thread_leaks(action="raise"):
            t = threading.Thread(target=lambda: time.sleep(10), name="db-poller")
            t.start()
            time.sleep(0.1)
    except ThreadLeakError as e:
        print(f"Thread leak: {e}")
```
```

--------------------------------

### Detect Synchronous HTTP Calls in Async Code

Source: https://context7.com/deepankarm/pyleak/llms.txt

Employ `no_event_loop_blocking` to identify synchronous network requests (e.g., using `requests`) within asynchronous code. The example also shows the async alternative using `httpx` which does not cause blocking.

```python
import asyncio
import time
import requests
from pyleak import no_event_loop_blocking

async def example_sync_http():
    async with no_event_loop_blocking(action="warn", threshold=0.05):
        # This synchronous call blocks the loop — will be flagged
        resp = requests.get("https://httpbin.org/get", timeout=5)

    # Async alternative — no blocking detected
    async with no_event_loop_blocking(action="warn", threshold=0.05):
        async with __import__("httpx").AsyncClient() as client:
            resp = await client.get("https://httpbin.org/get")
```

--------------------------------

### Load Test Commands

Source: https://github.com/deepankarm/pyleak/blob/main/examples/event_loop_detection/README.md

Commands to set up the environment, run tests, and execute a load test against the FastAPI application.

```bash
uv sync
docker-compose up -d   # Start MinIO

# Run tests
uv run pytest tests/ -v -s

# Run load test
uv run uvicorn pdf_ingest:app --reload
uv run python scripts/load_test.py scripts/sample.pdf 100
```

--------------------------------

### Include Creation Stack Trace for Leaked Tasks

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Shows how to enable the tracking of task creation stack traces by setting `enable_creation_tracking=True` in `no_task_leaks`. Note: This is not recommended for production due to potential side effects.

```python
async def main():
    try:
        async with no_task_leaks(action="raise", enable_creation_tracking=True):
            await leaky_function()
    except TaskLeakError as e:
        print(e)
```

--------------------------------

### Context Managers for Leak Detection

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Shows how to use pyleak detectors as context managers in both asynchronous and synchronous code.

```python
# AsyncIO tasks (async context)
async with no_task_leaks():
    # Your async code here
    pass

# Threads (sync context)
with no_thread_leaks():
    # Your threaded code here
    pass

# Event loop blocking (async context only)
async def main():
    async with no_event_loop_blocking():
        # Your potentially blocking code here
        pass
```

--------------------------------

### Configure Event Loop Blocking Detection

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Adjust `no_event_loop_blocking` settings, including action, logger, blocking time threshold, and check interval. The default action is 'warn'.

```python
no_event_loop_blocking(
    action="warn",           # Action to take on detection
    logger=None,             # Custom logger
    threshold=0.1,           # Minimum blocking time to report (seconds)
    check_interval=0.01      # How often to check (seconds)
)
```

--------------------------------

### Control Leak Detection Actions

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Demonstrates different actions ('cancel', 'raise', 'log') for `no_task_leaks`, `no_thread_leaks`, and `no_event_loop_blocking`. Choose the action that best suits your needs for handling detected leaks or blocking.

```python
# Examples
async with no_task_leaks(action="cancel"):  # Cancels leaked tasks
    pass

with no_thread_leaks(action="raise"):  # Raises exception on thread leaks
    pass

async with no_event_loop_blocking(action="log"):  # Logs blocking events
    pass
```

--------------------------------

### Detect Event Loop Blocking with pyleak

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Use `no_event_loop_blocking` as an async context manager to monitor for event loop blocking. Configure the `action` to 'raise' to immediately catch `EventLoopBlockError` when blocking exceeds the specified `threshold` (in seconds).

```python
import asyncio
from pyleak import EventLoopBlockError, no_event_loop_blocking

async def process_user_data(user_id: int):
    """Simulates cpu intensive work - contains blocking operations!"""
    print(f"Processing user {user_id}...")
    return sum(i * i for i in range(100_000_000))

async def main():
    try:
        async with no_event_loop_blocking(action="raise", threshold=0.5):
            user1 = await process_user_data(1)
            user2 = await process_user_data(2)

    except EventLoopBlockError as e:
        print(f"\n🚨 Found {e.block_count} blocking events:")
        print(e)

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

--------------------------------

### Decorators for Leak Detection

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Illustrates using pyleak detectors as decorators for functions and asynchronous functions.

```python
@no_task_leaks()
async def my_async_function():
    # Any leaked tasks will be detected
    pass

@no_thread_leaks()
def my_threaded_function():
    # Any leaked threads will be detected  
    pass

@no_event_loop_blocking()
async def my_potentially_blocking_function():
    # Any event loop blocking will be detected
    pass
```

--------------------------------

### Inspect LeakedTask fields with Pyleak context manager

Source: https://context7.com/deepankarm/pyleak/llms.txt

Use the no_task_leaks context manager with enable_creation_tracking=True to capture detailed information about leaked tasks, including name, ID, state, references, and stack traces. Leaked tasks can be cancelled.

```python
import asyncio
from pyleak import TaskLeakError, EventLoopBlockError, no_task_leaks, no_event_loop_blocking
import time

# ── Inspecting LeakedTask fields ──────────────────────────────────────────────
async def inspect_leaked_task():
    async def worker():
        await asyncio.sleep(60)

    try:
        async with no_task_leaks(action="raise", enable_creation_tracking=True):
            asyncio.create_task(worker(), name="my-worker")
            await asyncio.sleep(0.05)
    except TaskLeakError as e:
        task_info = e.leaked_tasks[0]
        print(task_info.name)          # "my-worker"
        print(task_info.task_id)       # numeric id(task)
        print(task_info.state)         # "running"
        print(task_info.task_ref)      # the live asyncio.Task

        print("=== Current Stack ===")
        print(task_info.format_current_stack())
        # File "/app/worker.py", line 3, in worker
        #     await asyncio.sleep(60)

        print("=== Creation Stack ===")
        print(task_info.format_creation_stack())
        # File "/app/main.py", line 9, in inspect_leaked_task
        #     asyncio.create_task(worker(), name="my-worker")

        # Cancel the leaked task
        if task_info.task_ref and not task_info.task_ref.done():
            task_info.task_ref.cancel()
```

--------------------------------

### Catching combined Pyleak exceptions programmatically

Source: https://context7.com/deepankarm/pyleak/llms.txt

Instantiate CombinedLeakDetector programmatically to use custom configurations for leak detection within a context manager. Caught PyleakExceptionGroup can be iterated to inspect individual errors.

```python
# ── Catching the combined exception group ─────────────────────────────────────
@pytest.mark.asyncio
async def test_catch_group():
    try:
        # manually instantiate CombinedLeakDetector for programmatic use
        from pyleak.combined import CombinedLeakDetector, PyLeakConfig
        config = PyLeakConfig(task_action="raise", blocking_action="raise")
        async with CombinedLeakDetector(config=config, is_async=True):
            asyncio.create_task(asyncio.sleep(10))
            time.sleep(0.5)
    except PyleakExceptionGroup as eg:
        for err in eg.exceptions:
            print(type(err).__name__, err)
```

--------------------------------

### Configure Thread Leak Detection

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Configure `no_thread_leaks` with an action, name filter, custom logger, and an option to exclude daemon threads. The default action is 'warn'.

```python
no_thread_leaks(
    action="warn",           # Action to take on detection
    name_filter=None,        # Filter by thread name
    logger=None,             # Custom logger
    exclude_daemon=True,     # Exclude daemon threads
)
```

--------------------------------

### FastAPI Endpoint with Offloaded Blocking Code

Source: https://github.com/deepankarm/pyleak/blob/main/examples/event_loop_detection/README.md

This FastAPI endpoint demonstrates the fix by offloading CPU-bound PDF processing to a thread pool using `asyncio.to_thread` and uploading images concurrently using `asyncio.gather`.

```python
from fastapi import FastAPI, UploadFile
import fitz
import asyncio

app = FastAPI()

def _process_pdf_sync(pdf_bytes):
    doc = fitz.open(stream=pdf_bytes, filetype="pdf")
    text = ""
    for page in doc:
        text += page.get_text()
    
    extracted = []
    for page_num in range(len(doc)):
        page = doc.load_page(page_num)
        img_list = page.get_images(full=True)
        for img_index, img in enumerate(img_list):
            xref = img[0]
            base_image = doc.extract_image(xref)
            image_bytes = base_image["image"]
            ext = base_image["ext"]
            extracted.append((page_num, img_index, image_bytes, ext))
    return text, len(doc), extracted

async def _upload_image_async(document_id, idx, page_num, img_index, img_bytes, ext):
    # Simulate async S3 upload
    import time
    import boto3
    s3_client = boto3.client("s3")
    s3_key = f"{document_id}/img_{idx}.{ext}"
    await asyncio.sleep(0.001) # Simulate async network I/O
    # s3_client.put_object(Bucket="your-bucket-name", Key=s3_key, Body=img_bytes, ContentType=f"image/{ext}")
    print(f"Uploaded {s3_key} async")
    return True

class IngestResponse:
    def __init__(self, **kwargs):
        self.__dict__.update(kwargs)

@app.post("/ingest/async")
async def ingest_async(file: UploadFile):
    pdf_bytes = await file.read()
    document_id = "doc_123"

    # Offload CPU-bound PDF processing to thread pool
    text, page_count, extracted = await asyncio.to_thread(_process_pdf_sync, pdf_bytes)

    # Upload images concurrently
    upload_tasks = [
        _upload_image_async(document_id, idx, page_num, img_index, img_bytes, ext)
        for idx, (page_num, img_index, img_bytes, ext) in enumerate(extracted)
    ]
    images = await asyncio.gather(*upload_tasks)

    return IngestResponse(message="File ingested successfully (async)")

```

--------------------------------

### Configure AsyncIO Task Leak Detection

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Customize `no_task_leaks` with an action, name filter, and a custom logger. The default action is 'warn'.

```python
no_task_leaks(
    action="warn",           # Action to take on detection
    name_filter=None,        # Filter by task name
    logger=None              # Custom logger
)
```

--------------------------------

### Tune Blocking Detection Threshold and Interval

Source: https://context7.com/deepankarm/pyleak/llms.txt

Customize `no_event_loop_blocking` by adjusting `threshold` for flagging blocking events and `check_interval` for the sampling frequency of the event loop monitor. This allows for fine-tuning detection sensitivity.

```python
import asyncio
import time
from pyleak import no_event_loop_blocking

async def example_tuned():
    async with no_event_loop_blocking(
        action="log",
        threshold=0.05,       # flag anything blocking > 50 ms
        check_interval=0.01,  # sample every 10 ms
    ):
        await asyncio.sleep(0.1)  # async → no block detected
```

--------------------------------

### Configure Pytest Markers in pyproject.toml

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Add the 'no_leaks' marker to your pytest configuration in pyproject.toml. This marker is used to enable pyleak's detection capabilities.

```toml
[tool.pytest.ini_options]
markers = [
    "no_leaks: detect asyncio task leaks, thread leaks, and event loop blocking"
]
```

--------------------------------

### Warn on Event Loop Blocking

Source: https://context7.com/deepankarm/pyleak/llms.txt

Use `no_event_loop_blocking` as an async context manager to detect and warn about synchronous code that blocks the event loop. Set a `threshold` to specify the maximum allowed blocking time.

```python
import asyncio
import time
from pyleak import no_event_loop_blocking

async def example_warn():
    async with no_event_loop_blocking(action="warn", threshold=0.1):
        time.sleep(0.5)  # blocks the event loop → ResourceWarning with stack trace
```

--------------------------------

### Custom actions for leak detectors with Pyleak marker

Source: https://context7.com/deepankarm/pyleak/llms.txt

Define custom actions ('raise', 'warn', 'ignore') for each detector type (task, thread, blocking) using the @pytest.mark.no_leaks decorator. A blocking threshold can also be specified.

```python
# ── Custom actions per detector ───────────────────────────────────────────────
@pytest.mark.no_leaks(
    task_action="raise",
    thread_action="warn",
    blocking_action="raise",
    blocking_threshold=0.1,
)
@pytest.mark.asyncio
async def test_custom_actions():
    await asyncio.sleep(0.01)
```

--------------------------------

### Configure Pytest Markers in pytest.ini

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Alternatively, configure the 'no_leaks' marker in your pytest.ini file. This marker enables pyleak's leak detection features.

```ini
[tool:pytest]
markers = no_leaks: detect asyncio task leaks, thread leaks, and event loop blocking
```

--------------------------------

### Grace Period for Thread Completion

Source: https://context7.com/deepankarm/pyleak/llms.txt

Configure `no_thread_leaks` with a `grace_period` to allow threads to finish within a specified time without reporting a leak. This is ideal for short-lived threads that are expected to complete before the context manager exits.

```python
import threading
import time
from pyleak import no_thread_leaks

def example_grace_period():
    with no_thread_leaks(action="raise", grace_period=0.5):
        t = threading.Thread(target=lambda: time.sleep(0.2))
        t.start()
        # thread completes within the 0.5 s grace period → no leak
```

--------------------------------

### Detect leaked threads with 'raise' action

Source: https://context7.com/deepankarm/pyleak/llms.txt

Configure `no_thread_leaks` to raise a `ThreadLeakError` upon detecting leaked threads. A `grace_period` parameter can be used to allow threads a short time to finish before checking.

```python
import threading
import time
from pyleak import ThreadLeakError, no_thread_leaks

# ── Raise exception on leak ────────────────────────────────────────────────────
def example_raise():
    try:
        with no_thread_leaks(action="raise"):
            t = threading.Thread(target=lambda: time.sleep(10), name="db-poller")
            t.start()
            time.sleep(0.1)
    except ThreadLeakError as e:
        print(f"Thread leak: {e}")
```

--------------------------------

### Detect leaked asyncio tasks with 'raise' action and creation tracking

Source: https://context7.com/deepankarm/pyleak/llms.txt

Configure `no_task_leaks` to raise a `TaskLeakError` upon detecting leaked tasks. Enabling `enable_creation_tracking=True` captures full stack traces for root-cause analysis. Leaked tasks can be optionally cancelled.

```python
import asyncio
from pyleak import TaskLeakError, no_task_leaks

# ── Raise exception with full stack traces ─────────────────────────────────────
async def example_raise():
    async def background():
        await asyncio.sleep(30)

    try:
        async with no_task_leaks(action="raise", enable_creation_tracking=True):
            asyncio.create_task(background(), name="worker-1")
            await asyncio.sleep(0.05)
    except TaskLeakError as e:
        print(f"Caught {e.task_count} leaked task(s)")
        for task in e.leaked_tasks:
            print(f"  Task: {task.name}, state: {task.state}")
            print("  Currently at:", task.format_current_stack())
            print("  Created at:  ", task.format_creation_stack())
            if task.task_ref and not task.task_ref.done():
                task.task_ref.cancel()
```

--------------------------------

### Detect leaked asyncio tasks with 'warn' action

Source: https://context7.com/deepankarm/pyleak/llms.txt

Use `no_task_leaks` as an async context manager to detect tasks that are created but never awaited or cancelled. This basic usage will issue a `ResourceWarning` if leaks are found.

```python
import asyncio
from pyleak import TaskLeakError, no_task_leaks

# ── Basic usage: warn on leak ──────────────────────────────────────────────────
async def example_warn():
    async with no_task_leaks(action="warn"):
        asyncio.create_task(asyncio.sleep(10))  # never awaited → ResourceWarning
        await asyncio.sleep(0.1)
```

--------------------------------

### FastAPI Endpoint with Blocking Code

Source: https://github.com/deepankarm/pyleak/blob/main/examples/event_loop_detection/README.md

This FastAPI endpoint uses synchronous libraries (PyMuPDF, boto3) for PDF processing and S3 uploads, which will block the asyncio event loop.

```python
from fastapi import FastAPI, UploadFile
import fitz  # PyMuPDF
import asyncio

app = FastAPI()

def _extract_text(doc):
    # Simulate blocking PDF processing
    text = ""
    for page in doc:
        text += page.get_text()
    return text

def _extract_images(doc):
    # Simulate blocking image extraction
    extracted = []
    for page_num in range(len(doc)):
        page = doc.load_page(page_num)
        img_list = page.get_images(full=True)
        for img_index, img in enumerate(img_list):
            xref = img[0]
            base_image = doc.extract_image(xref)
            image_bytes = base_image["image"]
            ext = base_image["ext"]
            extracted.append((page_num, img_index, image_bytes, ext))
    return extracted

def _upload_to_s3(s3_key, img_bytes, content_type):
    # Simulate blocking S3 upload
    import time
    import boto3
    s3_client = boto3.client("s3")
    time.sleep(0.01) # Simulate network latency and processing
    # s3_client.put_object(Bucket="your-bucket-name", Key=s3_key, Body=img_bytes, ContentType=content_type)
    print(f"Uploaded {s3_key}")
    return True

class IngestResponse:
    def __init__(self, **kwargs):
        self.__dict__.update(kwargs)

@app.post("/ingest/blocking")
async def ingest_blocking(file: UploadFile):
    pdf_bytes = await file.read()
    document_id = "doc_123"

    doc = fitz.open(stream=pdf_bytes, filetype="pdf")  # Blocks
    text = _extract_text(doc)  # Blocks
    extracted = _extract_images(doc)  # Blocks

    for idx, (page_num, img_index, img_bytes, ext) in enumerate(extracted):
        s3_key = f"{document_id}/img_{idx}.{ext}"
        _upload_to_s3(s3_key, img_bytes, f"image/{ext}")  # Blocks

    return IngestResponse(message="File ingested successfully (blocking)")

```

--------------------------------

### Add Pytest Marker in conftest.py

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Register the 'no_leaks' marker programmatically in your conftest.py file. This ensures the marker is recognized by pytest and pyleak.

```python
# conftest.py
import pytest

def pytest_configure(config):
    config.addinivalue_line(
        "markers", 
        "no_leaks: detect asyncio task leaks, thread leaks, and event loop blocking"
    )
```

--------------------------------

### Inspect EventLoopBlock Fields

Source: https://context7.com/deepankarm/pyleak/llms.txt

Catch `EventLoopBlockError` to access details about blocking events. This includes the total number of blocks, and specific information for each blocking event such as its ID, duration, threshold, timestamp, and the stack trace that caused the block. Use this to diagnose performance bottlenecks.

```python
async def inspect_block():
    try:
        async with no_event_loop_blocking(action="raise", threshold=0.1):
            time.sleep(0.5)
    except EventLoopBlockError as e:
        print(f"Total blocks: {e.block_count}")
        block = e.blocking_events[0]
        print(f"Block #{block.block_id}")
        print(f"Duration : {block.duration:.3f}s")
        print(f"Threshold: {block.threshold:.3f}s")
        print(f"Timestamp: {block.timestamp:.3f}")
        print("=== Blocking Stack ===")
        print(block.format_blocking_stack())
```

--------------------------------

### Ensure Proper Resource Cleanup with Async Tasks

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Utilize `no_task_leaks` to ensure that asynchronous tasks are properly cleaned up. The 'raise' action will fail the test if a task is not properly handled (e.g., cancelled and awaited).

```python
async def test_background_task_cleanup():
    async with no_task_leaks(action="raise"):
        # This would fail the test
        asyncio.create_task(long_running_task())
        
        # This would pass
        task = asyncio.create_task(long_running_task())
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass
```

--------------------------------

### Detect Synchronous HTTP Calls in Async Code

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Use `no_event_loop_blocking` to detect synchronous HTTP calls within asynchronous code. The 'warn' action will log a warning when a blocking call is made.

```python
import httpx
from starlette.testclient import TestClient

async def test_sync_vs_async_http():
    # This will detect blocking
    async with no_event_loop_blocking(action="warn"):
        response = TestClient(app).get("/endpoint")  # Synchronous!

    # This will not detect blocking
    async with no_event_loop_blocking(action="warn"):
        async with httpx.AsyncClient() as client:
            response = await client.get("/endpoint")  # Asynchronous!
```

--------------------------------

### Raise EventLoopBlockError on Blocking

Source: https://context7.com/deepankarm/pyleak/llms.txt

Configure `no_event_loop_blocking` with `action='raise'` to catch `EventLoopBlockError` when the event loop is blocked. The exception provides details about blocking events, including duration and stack traces.

```python
import asyncio
import time
from pyleak import EventLoopBlockError, no_event_loop_blocking

async def example_raise():
    async def compute():
        return sum(i * i for i in range(50_000_000))  # CPU-bound

    try:
        async with no_event_loop_blocking(action="raise", threshold=0.2):
            result = await compute()
    except EventLoopBlockError as e:
        print(f"{e.block_count} blocking event(s) detected")
        for block in e.blocking_events:
            print(f
```

--------------------------------

### Basic Usage of no_leaks Marker

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Apply the 'no_leaks' marker to an async test function to automatically detect asyncio task leaks. Ensure the 'asyncio' marker is also present for async tests.

```python
@pytest.mark.no_leaks
@pytest.mark.asyncio
async def test_no_task_leaks():
    asyncio.create_task(asyncio.sleep(10))
```

--------------------------------

### Enable all detectors with Pyleak marker

Source: https://context7.com/deepankarm/pyleak/llms.txt

Use the @pytest.mark.no_leaks decorator to enable all leak detection (tasks, threads, blocking) by default. Any detected leak will raise a PyleakExceptionGroup.

```python
import asyncio
import threading
import time
import pytest
from pyleak import PyleakExceptionGroup

# ── Enable all detectors (default) ────────────────────────────────────────────
@pytest.mark.no_leaks
@pytest.mark.asyncio
async def test_all_detectors():
    # Any task leak, thread leak, or event loop block raises PyleakExceptionGroup
    await asyncio.sleep(0.01)
```

--------------------------------

### Testing Blocking Endpoint Detection with pyleak

Source: https://github.com/deepankarm/pyleak/blob/main/examples/event_loop_detection/README.md

This pytest test uses pyleak's `no_event_loop_blocking` context manager to assert that the `/ingest/blocking` endpoint correctly triggers a blocking event loop detection.

```python
import pytest
from httpx import AsyncClient
from asgiref.testing import ASGITransport
from pyleak import no_event_loop_blocking

# Assuming 'app' is your FastAPI application instance from the previous example
# from your_app_module import app

# Mock app for demonstration if not imported
from fastapi import FastAPI, UploadFile
import fitz
import asyncio

app = FastAPI()

def _extract_text(doc):
    text = ""
    for page in doc:
        text += page.get_text()
    return text

def _extract_images(doc):
    extracted = []
    for page_num in range(len(doc)):
        page = doc.load_page(page_num)
        img_list = page.get_images(full=True)
        for img_index, img in enumerate(img_list):
            xref = img[0]
            base_image = doc.extract_image(xref)
            image_bytes = base_image["image"]
            ext = base_image["ext"]
            extracted.append((page_num, img_index, image_bytes, ext))
    return extracted

def _upload_to_s3(s3_key, img_bytes, content_type):
    import time
    import boto3
    s3_client = boto3.client("s3")
    time.sleep(0.01)
    print(f"Uploaded {s3_key}")
    return True

class IngestResponse:
    def __init__(self, **kwargs):
        self.__dict__.update(kwargs)

@app.post("/ingest/blocking")
async def ingest_blocking(file: UploadFile):
    pdf_bytes = await file.read()
    document_id = "doc_123"

    doc = fitz.open(stream=pdf_bytes, filetype="pdf")
    text = _extract_text(doc)
    extracted = _extract_images(doc)

    for idx, (page_num, img_index, img_bytes, ext) in enumerate(extracted):
        s3_key = f"{document_id}/img_{idx}.{ext}"
        _upload_to_s3(s3_key, img_bytes, f"image/{ext}")

    return IngestResponse(message="File ingested successfully (blocking)")


@pytest.mark.asyncio
async def test_blocking_endpoint_detected(sample_pdf: bytes):
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        with no_event_loop_blocking(action="raise", threshold=0.01):
            await client.post("/ingest/blocking", files={"file": ("test.pdf", sample_pdf)})

```

--------------------------------

### Debug Task Leaks in Asyncio

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

This traceback shows a typical scenario where task leaks might occur in an asyncio application. It highlights the call stack leading to the creation of a worker task.

```python
handle._run()
  File "/opt/homebrew/anaconda3/envs/ffa/lib/python3.12/asyncio/events.py", line 84, in _run
    self._context.run(self._callback, *self._args)
  File "/private/tmp/example.py", line 47, in debug_task_leaks
    await spawn_workers()
  File "/private/tmp/example.py", line 39, in spawn_workers
    asyncio.create_task(worker(i, random.randint(1, 10)), name=f"worker-{i}")
```

--------------------------------

### Warn on Daemon Threads with no_thread_leaks

Source: https://context7.com/deepankarm/pyleak/llms.txt

Use `no_thread_leaks` with `exclude_daemon=False` to warn about daemon threads that might otherwise be ignored. This is useful for ensuring all threads, including daemon threads, are properly managed.

```python
import threading
import time
from pyleak import no_thread_leaks

def example_daemon():
    with no_thread_leaks(action="warn", exclude_daemon=False):
        t = threading.Thread(target=lambda: time.sleep(10), name="cache-refresh")
        t.daemon = True
        t.start()
        time.sleep(0.1)  # detected because exclude_daemon=False
```

--------------------------------

### Selective leak detection with Pyleak marker

Source: https://context7.com/deepankarm/pyleak/llms.txt

Configure the @pytest.mark.no_leaks decorator to enable only specific detectors (e.g., tasks and blocking) and disable others (e.g., threads). Leaks in enabled detectors will raise exceptions.

```python
# ── Selective detection ────────────────────────────────────────────────────────
@pytest.mark.no_leaks(tasks=True, blocking=True, threads=False)
@pytest.mark.asyncio
async def test_tasks_and_blocking_only():
    asyncio.create_task(asyncio.sleep(10))  # detected → raises
    time.sleep(0.5)                         # detected → raises
    threading.Thread(target=lambda: time.sleep(10)).start()  # ignored
```

--------------------------------

### no_task_leaks

Source: https://context7.com/deepankarm/pyleak/llms.txt

Detects leaked asyncio tasks that were created but never awaited or cancelled. It can issue warnings, raise exceptions with stack traces, or automatically cancel leaked tasks. Supports filtering by task name using exact strings or regex, and can track task creation stacks.

```APIDOC
## no_task_leaks

### Description
Wraps async code to compare the set of running asyncio tasks before and after execution. Reports tasks created within the block that are still running and not awaited or cancelled as leaks.

### Usage
Can be used as an async context manager or a function decorator.

### Parameters
- **action** (str): The action to take upon detecting a leak. Options include "warn", "raise", or "cancel".
- **enable_creation_tracking** (bool, optional): If True, enables asyncio debug mode to capture the full stack trace at the point each task was created. Defaults to False.
- **name_filter** (re.Pattern, optional): A regex pattern to filter tasks by name. Only tasks matching this pattern will be considered for leak detection.

### Actions
- **warn**: Issues a `ResourceWarning`.
- **raise**: Raises a `TaskLeakError` with details about the leaked tasks and their stack traces.
- **cancel**: Automatically cancels leaked tasks.

### Examples

#### Basic usage: warn on leak
```python
import asyncio
from pyleak import no_task_leaks

async def example_warn():
    async with no_task_leaks(action="warn"):
        asyncio.create_task(asyncio.sleep(10))  # never awaited → ResourceWarning
        await asyncio.sleep(0.1)
```

#### Raise exception with full stack traces
```python
import asyncio
from pyleak import TaskLeakError, no_task_leaks

async def example_raise():
    async def background():
        await asyncio.sleep(30)

    try:
        async with no_task_leaks(action="raise", enable_creation_tracking=True):
            asyncio.create_task(background(), name="worker-1")
            await asyncio.sleep(0.05)
    except TaskLeakError as e:
        print(f"Caught {e.task_count} leaked task(s)")
        for task in e.leaked_tasks:
            print(f"  Task: {task.name}, state: {task.state}")
            print("  Currently at:", task.format_current_stack())
            print("  Created at:  ", task.format_creation_stack())
            if task.task_ref and not task.task_ref.done():
                task.task_ref.cancel()
```

#### Cancel leaked tasks automatically
```python
import asyncio
from pyleak import no_task_leaks

async def example_cancel():
    async with no_task_leaks(action="cancel"):
        asyncio.create_task(asyncio.sleep(100))  # will be cancelled on exit
        await asyncio.sleep(0.05)
```

#### Filter by name using regex
```python
import asyncio
import re
from pyleak import no_task_leaks

async def example_filter():
    async with no_task_leaks(action="raise", name_filter=re.compile(r"worker-\d+")):
        asyncio.create_task(asyncio.sleep(10), name="worker-1")   # detected
        asyncio.create_task(asyncio.sleep(10), name="heartbeat")  # ignored
        await asyncio.sleep(0.05)
```

#### As a decorator
```python
import asyncio
from pyleak import no_task_leaks

@no_task_leaks(action="raise", enable_creation_tracking=True)
async def my_service_handler():
    asyncio.create_task(asyncio.sleep(60))  # TaskLeakError raised after function returns
    await asyncio.sleep(0.1)
```
```

--------------------------------

### Filter Leaks by Name

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Use `name_filter` with exact strings or regex patterns to specify which tasks or threads to monitor. Note that event loop blocking detection does not support name filtering.

```python
import re

# Exact match
async with no_task_leaks(name_filter="background-worker"):
    pass

with no_thread_leaks(name_filter="worker-thread"):
    pass

# Regex pattern
async with no_task_leaks(name_filter=re.compile(r"worker-\d+")):
    pass

with no_thread_leaks(name_filter=re.compile(r"background-.*\))
    pass
```

--------------------------------

### Pytest Plugin for Leak Detection

Source: https://context7.com/deepankarm/pyleak/llms.txt

The `@pytest.mark.no_leaks` marker integrates pyleak detectors into pytest tests. It automatically wraps marked tests with `CombinedLeakDetector` for sync and async tests, grouping errors into `PyleakExceptionGroup`.

```python
# pyproject.toml (register marker to silence PytestUnknownMarkWarning)
# [tool.pytest.ini_options]

```

--------------------------------

### Filter Threads by Regex Name

Source: https://context7.com/deepankarm/pyleak/llms.txt

Utilize the `name_filter` argument in `no_thread_leaks` with a compiled regex to selectively monitor threads based on their names. This allows for fine-grained control over which threads trigger leak warnings.

```python
import re
import threading
import time
from pyleak import no_thread_leaks

def example_filter():
    with no_thread_leaks(action="raise", name_filter=re.compile(r"^pool-worker-")):
        threading.Thread(target=lambda: time.sleep(10), name="pool-worker-1").start()
        threading.Thread(target=lambda: time.sleep(10), name="logger").start()
        time.sleep(0.1)  # only "pool-worker-1" raises
```

--------------------------------

### Decorator for Async Event Loop Blocking Detection

Source: https://context7.com/deepankarm/pyleak/llms.txt

Use `no_event_loop_blocking` as a decorator on async functions to automatically monitor for event loop blocking. This is convenient for ensuring that entire async operations do not inadvertently block the event loop.

```python
import asyncio
import time

@no_event_loop_blocking(action="raise", threshold=0.1)
async def fetch_data(url: str):
    import httpx
    async with httpx.AsyncClient() as client:
        return await client.get(url)  # fully async → no issue

@no_event_loop_blocking(action="warn")
async def bad_handler():
    time.sleep(2)  # ResourceWarning issued after function returns
```

--------------------------------

### Debug Complex Task Leaks in Python

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Use `no_task_leaks` to detect and report leaked asyncio tasks. Configure `action='raise'` to catch `TaskLeakError` and `enable_creation_tracking=True` for detailed stack traces. A `name_filter` can be used to focus on specific task patterns.

```python
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())

```

--------------------------------

### Filter tasks by name with Pyleak marker

Source: https://context7.com/deepankarm/pyleak/llms.txt

Use the 'task_name_filter' argument in the @pytest.mark.no_leaks decorator with a regex to specify which tasks should be monitored for leaks. Tasks not matching the filter are ignored.

```python
# ── Task name filter in marker ────────────────────────────────────────────────
@pytest.mark.no_leaks(task_name_filter=r"^background-")
@pytest.mark.asyncio
async def test_filtered_tasks():
    asyncio.create_task(asyncio.sleep(10), name="background-job")  # caught
    asyncio.create_task(asyncio.sleep(10), name="system-heartbeat")  # ignored
    await asyncio.sleep(0.05)
```

--------------------------------

### Test No Leaked Threads

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Use `no_thread_leaks(action='raise')` within a pytest test function to verify that no threads are leaked. This is suitable for synchronous code.

```python
def test_no_leaked_threads():
    with no_thread_leaks(action="raise"):
        my_threaded_function()
```

--------------------------------

### Automatically cancel leaked asyncio tasks

Source: https://context7.com/deepankarm/pyleak/llms.txt

Use `no_task_leaks` with `action="cancel"` to automatically cancel any detected leaked asyncio tasks when the context manager exits. This prevents them from running indefinitely.

```python
import asyncio
from pyleak import TaskLeakError, no_task_leaks

# ── Cancel leaked tasks automatically ─────────────────────────────────────────
async def example_cancel():
    async with no_task_leaks(action="cancel"):
        asyncio.create_task(asyncio.sleep(100))  # will be cancelled on exit
        await asyncio.sleep(0.05)
```

--------------------------------

### Filter asyncio task leaks by name using regex

Source: https://context7.com/deepankarm/pyleak/llms.txt

Employ `no_task_leaks` with a `name_filter` (a compiled regex object) to selectively detect leaks only from tasks matching the specified pattern. Other tasks will be ignored.

```python
import asyncio
from pyleak import TaskLeakError, no_task_leaks
import re

# ── Filter by name using regex ─────────────────────────────────────────────────
async def example_filter():
    async with no_task_leaks(action="raise", name_filter=re.compile(r"worker-\d+")):
        asyncio.create_task(asyncio.sleep(10), name="worker-1")   # detected
        asyncio.create_task(asyncio.sleep(10), name="heartbeat")  # ignored
        await asyncio.sleep(0.05)
```

--------------------------------

### Test No Leaked Tasks

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Use `no_task_leaks(action='raise')` within a pytest test function to ensure no asyncio tasks are leaked. This requires the `pytest-asyncio` plugin.

```python
@pytest.mark.asyncio
async def test_no_leaked_tasks():
    async with no_task_leaks(action="raise"):
        await my_async_function()
```

--------------------------------

### Use `no_task_leaks` as a decorator

Source: https://context7.com/deepankarm/pyleak/llms.txt

Apply `no_task_leaks` as a decorator to an async function to automatically wrap its execution and detect leaked tasks. This is useful for ensuring cleanup in service handlers or other async routines.

```python
import asyncio
from pyleak import TaskLeakError, no_task_leaks

# ── As a decorator ─────────────────────────────────────────────────────────────
@no_task_leaks(action="raise", enable_creation_tracking=True)
async def my_service_handler():
    asyncio.create_task(asyncio.sleep(60))  # TaskLeakError raised after function returns
    await asyncio.sleep(0.1)
```

--------------------------------

### Decorator for Thread Leak Detection

Source: https://context7.com/deepankarm/pyleak/llms.txt

Apply `no_thread_leaks` as a decorator to functions to automatically detect leaked threads upon function exit. This simplifies leak detection for entire functions or methods.

```python
import threading
import time
from pyleak import no_thread_leaks

@no_thread_leaks(action="warn")
def process_batch(items):
    for item in items:
        t = threading.Thread(target=lambda: time.sleep(5))
        t.start()  # leaked threads → ResourceWarning after function returns
```

--------------------------------

### Synchronous test for thread leak detection

Source: https://context7.com/deepankarm/pyleak/llms.txt

Configure the @pytest.mark.no_leaks decorator to specifically enable thread leak detection for synchronous test functions. Other detectors can be disabled.

```python
# ── Sync test with thread leak detection ──────────────────────────────────────
@pytest.mark.no_leaks(tasks=False, blocking=False, threads=True)
def test_sync_thread_leak():
    t = threading.Thread(target=lambda: time.sleep(10), name="leaked")
    t.start()
    time.sleep(0.1)  # → ThreadLeakError raised
```

--------------------------------

### Test No Event Loop Blocking

Source: https://github.com/deepankarm/pyleak/blob/main/README.md

Employ `no_event_loop_blocking(action='raise', threshold=0.1)` in pytest tests to detect and raise errors for event loop blocking. This is useful for testing potentially blocking asynchronous code.

```python
@pytest.mark.asyncio        
async def test_no_event_loop_blocking():
    async with no_event_loop_blocking(action="raise", threshold=0.1):
        await my_potentially_blocking_function()
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.