Growth Analysis and Monitoring

def growth(limit=10, peak_stats={}, shortnames=True, filter=None):
    """
    Count the increase in peak object counts since last call.
    
    Args:
        limit (int): Maximum number of deltas to return; None to see all
        peak_stats (dict): Dictionary from type names to previously seen peak counts; usually managed automatically
        shortnames (bool): If True, use short class names; if False, use fully qualified names
        filter (callable, optional): Function taking object and returning bool; objects returning False are ignored
        
    Returns:
        list: List of (type_name, total_count, increase_delta) tuples, descending by increase_delta
        
    Note:
        Calls gc.collect() automatically and updates peak_stats with new peak values.
    """

import objgraph

# Initial measurement - establishes baseline
objgraph.growth()

# ... run some code that might create objects ...

# Check for growth since last call
growth_data = objgraph.growth()
for type_name, total, delta in growth_data:
    print(f"{type_name}: {total} total ({delta:+d} since last check)")

# Use persistent peak stats across calls
peak_stats = {}
initial_growth = objgraph.growth(peak_stats=peak_stats)
# ... more code ...
later_growth = objgraph.growth(peak_stats=peak_stats)

# Get unlimited results
all_growth = objgraph.growth(limit=None)

def show_growth(limit=10, peak_stats=None, shortnames=True, file=None, filter=None):
    """
    Show the increase in peak object counts since last call.
    
    Args:
        limit (int): Maximum number of entries to show
        peak_stats (dict, optional): Peak stats dictionary; if None, uses internal state
        shortnames (bool): If True, use short class names; if False, use fully qualified names
        file (file-like, optional): Output destination; defaults to sys.stdout  
        filter (callable, optional): Function taking object and returning bool; objects returning False are ignored
        
    Returns:
        None: Prints formatted output
    """

import objgraph
import sys

# Basic growth monitoring workflow
objgraph.show_growth()  # Establish baseline
# ... run some code ...
objgraph.show_growth()  # Show changes

# Custom file output
with open('growth_log.txt', 'w') as f:
    objgraph.show_growth(file=f)

# Use external peak stats for more control
peak_stats = {}
objgraph.show_growth(peak_stats=peak_stats)
# ... later ...
objgraph.show_growth(peak_stats=peak_stats)

# Show fully qualified names
objgraph.show_growth(shortnames=False)

# Filter to specific object types
def is_my_class(obj):
    return type(obj).__name__.startswith('My')

objgraph.show_growth(filter=is_my_class)

wrapper_descriptor       970       +14
tuple                  12282       +10  
dict                    1922        +7
list                    1433        +3

def get_new_ids(skip_update=False, limit=10, sortby='deltas', shortnames=None, file=None, _state={}):
    """
    Find and display new objects allocated since last call.
    
    Args:
        skip_update (bool): If True, returns previous results without updating internal state
        limit (int): Maximum number of rows to print; 0 to suppress printing; None to print everything
        sortby (str): Column to sort by ('old', 'current', 'new', 'deltas')
        shortnames (bool, optional): Use short class names; if None, remembers previous setting (default True)
        file (file-like, optional): Output destination; defaults to sys.stdout
        _state (dict): Internal state storage; do not pass manually
        
    Returns:
        dict: Mapping from type names to sets of object IDs created since last call
        
    Note:
        Calls gc.collect() automatically. Use with at_addrs() to get actual objects from IDs.
    """

import objgraph

# Establish baseline
objgraph.get_new_ids(limit=0)  # Don't print, just establish state

# ... create some objects ...
a = [1, 2, 3]
b = {'key': 'value'}
c = [4, 5, 6]

# Get new object IDs and print summary
new_ids = objgraph.get_new_ids()

# Get actual objects from IDs
new_lists = objgraph.at_addrs(new_ids.get('list', set()))
print(f"New lists created: {len(new_lists)}")
print(f"Our list 'a' is in new lists: {a in new_lists}")

# Skip update to get previous results again
same_ids = objgraph.get_new_ids(skip_update=True)

# Sort by different columns
objgraph.get_new_ids(sortby='current')  # Sort by current count
objgraph.get_new_ids(sortby='new')      # Sort by new objects count
objgraph.get_new_ids(sortby='old')      # Sort by old count

# Unlimited output
all_new_ids = objgraph.get_new_ids(limit=None)

# Fully qualified names
objgraph.get_new_ids(shortnames=False)

======================================================================
Type                    Old_ids  Current_ids      New_ids Count_Deltas
======================================================================  
list                        324          326           +3           +2
dict                       1125         1125           +0           +0
wrapper_descriptor         1001         1001           +0           +0
======================================================================

import objgraph

# Establish baseline
objgraph.show_growth()

# Run your code that might leak memory
for i in range(1000):
    # some_potentially_leaky_operation()
    pass

# Check for consistent growth  
objgraph.show_growth()

# Get specific objects for investigation
new_ids = objgraph.get_new_ids()
if 'MyClass' in new_ids:
    leaked_objects = objgraph.at_addrs(new_ids['MyClass'])
    # Investigate what's keeping these objects alive
    objgraph.show_backrefs(leaked_objects[:5])

import objgraph
import time

peak_stats = {}

while True:
    objgraph.show_growth(peak_stats=peak_stats)
    time.sleep(60)  # Check every minute