Memory Management in Python: A Comprehensive Guide

# When you create objects, Python automatically allocates memory my_list = [1, 2, 3, 4, 5] # Memory allocated for list object my_dict = {"name": "John", "age": 30} # Memory allocated for dict object my_string = "Hello, World!" # Memory allocated for string object  # You can check memory address print(id(my_list)) # e.g., 140234567890123 print(id(my_dict)) # e.g., 140234567890456 print(id(my_string)) # e.g., 140234567890789 

import sys # Create an object x = [1, 2, 3] print(sys.getrefcount(x)) # 2 (x + temporary reference in getrefcount)  # Create another reference y = x print(sys.getrefcount(x)) # 3 (x, y + temporary reference)  # Delete a reference del y print(sys.getrefcount(x)) # 2 (back to just x + temporary)  # When refcount reaches 0, memory is freed del x # Object is deallocated 

import gc # Circular reference example class Node: def __init__(self, value): self.value = value self.ref = None # Create circular reference node1 = Node(1) node2 = Node(2) node1.ref = node2 node2.ref = node1 # Circular reference!  # Even after deleting references, objects exist due to circular ref del node1 del node2 # Check garbage collector stats print(gc.get_count()) # (threshold0, threshold1, threshold2)  # Force garbage collection collected = gc.collect() print(f"Garbage collector freed {collected} objects") 

# Small objects (< 512 bytes) use PyMalloc small_list = [1, 2, 3] # Uses PyMalloc small_string = "Hello" # Uses PyMalloc  # Larger objects use system malloc large_list = list(range(100000)) # Uses system malloc 

# Python pools example - strings a = "hello" b = "hello" print(a is b) # True - Python interns small strings  # But not for all strings c = "hello world!" d = "hello world!" print(c is d) # False - not interned  # Numbers -5 to 256 are pre-allocated x = 100 y = 100 print(x is y) # True - same object  x = 1000 y = 1000 print(x is y) # False - different objects 

# Lists over-allocate for efficiency import sys my_list = [] print(sys.getsizeof(my_list)) # Empty list size  for i in range(10): my_list.append(i) print(f"Length: {len(my_list)}, Size: {sys.getsizeof(my_list)} bytes") # Notice size increases in chunks, not linearly 

# Install: pip install memory-profiler  from memory_profiler import profile @profile def memory_intensive_function(): # Create large list  big_list = [i for i in range(1000000)] # Create dictionary  big_dict = {i: i**2 for i in range(100000)} # Delete to free memory  del big_list del big_dict return "Done" # Run with: python -m memory_profiler script.py 

import tracemalloc # Start tracing tracemalloc.start() # Your code here data = [i for i in range(1000000)] more_data = {i: i**2 for i in range(100000)} # Get current memory usage current, peak = tracemalloc.get_traced_memory() print(f"Current memory usage: {current / 1024 / 1024:.2f} MB") print(f"Peak memory usage: {peak / 1024 / 1024:.2f} MB") # Get top memory allocations snapshot = tracemalloc.take_snapshot() top_stats = snapshot.statistics('lineno') print("[ Top 10 ]") for stat in top_stats[:10]: print(stat) tracemalloc.stop() 

import gc # Enable garbage collection debugging gc.set_debug(gc.DEBUG_STATS | gc.DEBUG_LEAK) # Track objects class TrackedObject: def __init__(self, name): self.name = name def __del__(self): print(f"Deleting {self.name}") # Create objects obj1 = TrackedObject("Object 1") obj2 = TrackedObject("Object 2") # Create circular reference obj1.ref = obj2 obj2.ref = obj1 # Delete references del obj1 del obj2 # Force collection to see debug output gc.collect() 

# Common memory leak - holding references in global containers cache = {} def process_data(key, data): # This keeps growing without bound!  cache[key] = expensive_computation(data) return cache[key] # Solution 1: Use weak references import weakref class Cache: def __init__(self): self._cache = weakref.WeakValueDictionary() def get(self, key, compute_func, *args): if key in self._cache: return self._cache[key] value = compute_func(*args) self._cache[key] = value return value # Solution 2: Implement LRU cache from functools import lru_cache @lru_cache(maxsize=128) def expensive_computation(data): # Automatically manages cache size  return data ** 2 

# Inefficient - creates intermediate lists def inefficient_processing(): data = list(range(10000000)) # Large list in memory  squared = [x**2 for x in data] # Another large list  filtered = [x for x in squared if x % 2 == 0] # Yet another!  return sum(filtered) # Efficient - uses generators def efficient_processing(): data = range(10000000) # No list, just range object  squared = (x**2 for x in data) # Generator, no memory  filtered = (x for x in squared if x % 2 == 0) # Generator  return sum(filtered) # Only one value in memory at a time  # Memory comparison import sys list_comp = [x**2 for x in range(1000)] gen_exp = (x**2 for x in range(1000)) print(f"List size: {sys.getsizeof(list_comp)} bytes") print(f"Generator size: {sys.getsizeof(gen_exp)} bytes") 

# Inefficient string concatenation def bad_string_concat(n): result = "" for i in range(n): result += str(i) # Creates new string object each time  return result # Efficient approaches def good_string_concat(n): # Using join  return ''.join(str(i) for i in range(n)) def better_string_concat(n): # Using StringIO  from io import StringIO buffer = StringIO() for i in range(n): buffer.write(str(i)) return buffer.getvalue() # Performance test import time n = 50000 start = time.time() bad_string_concat(n) print(f"Bad method: {time.time() - start:.2f} seconds") start = time.time() good_string_concat(n) print(f"Good method: {time.time() - start:.2f} seconds") 

# Automatic resource cleanup class LargeDataProcessor: def __init__(self, filename): self.filename = filename self.data = None def __enter__(self): print(f"Loading data from {self.filename}") self.data = self._load_large_file() return self def __exit__(self, exc_type, exc_val, exc_tb): print("Cleaning up resources") del self.data gc.collect() # Force garbage collection  def _load_large_file(self): # Simulate loading large file  return [i for i in range(1000000)] def process(self): return sum(self.data) / len(self.data) # Usage with LargeDataProcessor('data.txt') as processor: result = processor.process() print(f"Result: {result}") # Memory automatically cleaned up here 

# Without __slots__ class RegularClass: def __init__(self, x, y): self.x = x self.y = y # With __slots__ class SlottedClass: __slots__ = ['x', 'y'] def __init__(self, x, y): self.x = x self.y = y # Memory comparison import sys regular = RegularClass(1, 2) slotted = SlottedClass(1, 2) print(f"Regular class instance: {sys.getsizeof(regular.__dict__)} bytes (dict)") print(f"Slotted class instance: {sys.getsizeof(slotted)} bytes") # Creating many instances regular_list = [RegularClass(i, i) for i in range(10000)] slotted_list = [SlottedClass(i, i) for i in range(10000)] # Slotted version uses significantly less memory 

def process_large_dataset(): # Load large dataset  huge_data = load_gigabytes_of_data() # Process it  result = analyze_data(huge_data) # Explicitly delete when done  del huge_data # Force garbage collection if needed  gc.collect() # Continue with result  return result 

import psutil import os def get_memory_info(): """Get current process memory information""" process = psutil.Process(os.getpid()) memory_info = process.memory_info() return { 'rss': memory_info.rss / 1024 / 1024, # Resident Set Size in MB  'vms': memory_info.vms / 1024 / 1024, # Virtual Memory Size in MB  'percent': process.memory_percent(), 'available': psutil.virtual_memory().available / 1024 / 1024 } # Memory monitoring decorator def monitor_memory(func): def wrapper(*args, **kwargs): # Before execution  before = get_memory_info() print(f"Memory before {func.__name__}: {before['rss']:.2f} MB") # Execute function  result = func(*args, **kwargs) # After execution  after = get_memory_info() print(f"Memory after {func.__name__}: {after['rss']:.2f} MB") print(f"Memory increase: {after['rss'] - before['rss']:.2f} MB") return result return wrapper # Example usage @monitor_memory def memory_intensive_operation(): data = [i ** 2 for i in range(1000000)] processed = sorted(data, reverse=True) return len(processed) result = memory_intensive_operation() 

import threading import time import logging from collections import deque class MemoryMonitor: def __init__(self, threshold_mb=500, check_interval=60): self.threshold_mb = threshold_mb self.check_interval = check_interval self.memory_history = deque(maxlen=100) self.monitoring = False self.logger = logging.getLogger(__name__) def start_monitoring(self): """Start background memory monitoring""" self.monitoring = True monitor_thread = threading.Thread(target=self._monitor_loop) monitor_thread.daemon = True monitor_thread.start() def stop_monitoring(self): """Stop memory monitoring""" self.monitoring = False def _monitor_loop(self): """Background monitoring loop""" while self.monitoring: memory_info = get_memory_info() self.memory_history.append({ 'timestamp': time.time(), 'rss_mb': memory_info['rss'], 'percent': memory_info['percent'] }) # Check for memory threshold  if memory_info['rss'] > self.threshold_mb: self._handle_high_memory(memory_info) time.sleep(self.check_interval) def _handle_high_memory(self, memory_info): """Handle high memory usage""" self.logger.warning( f"High memory usage detected: {memory_info['rss']:.2f} MB " f"({memory_info['percent']:.1f}%)" ) # Trigger garbage collection  import gc collected = gc.collect() self.logger.info(f"Garbage collector freed {collected} objects") # Log memory allocations  self._log_top_allocations() def _log_top_allocations(self): """Log top memory allocations using tracemalloc""" import tracemalloc if not tracemalloc.is_tracing(): return snapshot = tracemalloc.take_snapshot() top_stats = snapshot.statistics('lineno') self.logger.info("Top memory allocations:") for index, stat in enumerate(top_stats[:5], 1): self.logger.info(f"{index}. {stat}") def get_memory_trend(self): """Analyze memory usage trend""" if len(self.memory_history) < 2: return "insufficient_data" recent_memory = [h['rss_mb'] for h in list(self.memory_history)[-10:]] avg_recent = sum(recent_memory) / len(recent_memory) older_memory = [h['rss_mb'] for h in list(self.memory_history)[-20:-10]] if older_memory: avg_older = sum(older_memory) / len(older_memory) if avg_recent > avg_older * 1.2: return "increasing" elif avg_recent < avg_older * 0.8: return "decreasing" return "stable" # Usage in production monitor = MemoryMonitor(threshold_mb=512, check_interval=30) monitor.start_monitoring() 

import mmap import os def process_large_file_efficiently(filename): """Process large file using memory mapping""" file_size = os.path.getsize(filename) with open(filename, 'r+b') as f: # Memory-map the file  with mmap.mmap(f.fileno(), file_size) as mmapped_file: # File is not loaded into memory until accessed  # Process in chunks  chunk_size = 1024 * 1024 # 1MB chunks  for i in range(0, file_size, chunk_size): chunk = mmapped_file[i:i + chunk_size] # Process chunk  process_chunk(chunk) # Memory is automatically released  def process_chunk(chunk): """Process a chunk of data""" # Your processing logic here  pass 

class ObjectPool: """Reuse objects to reduce memory allocation overhead""" def __init__(self, create_func, reset_func, max_size=100): self.create_func = create_func self.reset_func = reset_func self.max_size = max_size self._available = [] self._in_use = set() def acquire(self): """Get an object from the pool""" if self._available: obj = self._available.pop() else: obj = self.create_func() self._in_use.add(obj) return obj def release(self, obj): """Return an object to the pool""" if obj in self._in_use: self._in_use.remove(obj) self.reset_func(obj) if len(self._available) < self.max_size: self._available.append(obj) # else: let it be garbage collected  # Example: Connection pool def create_connection(): return {'connected': True, 'data': None} def reset_connection(conn): conn['data'] = None pool = ObjectPool(create_connection, reset_connection, max_size=10) # Use connections from pool conn1 = pool.acquire() # ... use connection ... pool.release(conn1) # Reused instead of destroyed 

import weakref import gc class CachedObject: """Object that can be cached with weak references""" _cache = weakref.WeakValueDictionary() def __new__(cls, key): # Check if object already exists in cache  obj = cls._cache.get(key) if obj is not None: return obj # Create new object  obj = super().__new__(cls) cls._cache[key] = obj return obj def __init__(self, key): self.key = key self.data = f"Data for {key}" # Example usage obj1 = CachedObject("key1") obj2 = CachedObject("key1") # Returns same object print(obj1 is obj2) # True  # When all strong references are gone, object is removed from cache del obj1 del obj2 gc.collect() obj3 = CachedObject("key1") # Creates new object 

class LazyDataLoader: """Load data only when accessed""" def __init__(self, data_source): self.data_source = data_source self._data = None @property def data(self): if self._data is None: print(f"Loading data from {self.data_source}") self._data = self._load_data() return self._data def _load_data(self): # Simulate expensive data loading  return [i ** 2 for i in range(1000000)] def clear_cache(self): """Explicitly clear cached data""" self._data = None gc.collect() # Usage loader = LazyDataLoader("database") # Data not loaded yet print("Object created") # Data loaded on first access result = sum(loader.data) # Triggers loading print(f"Sum: {result}") # Subsequent access uses cached data result2 = sum(loader.data) # No loading  # Clear when done loader.clear_cache() 

def process_large_csv(filename): """Process large CSV file without loading everything into memory""" import csv def process_batch(batch): # Process batch of rows  return sum(float(row['value']) for row in batch) batch_size = 1000 batch = [] total = 0 with open(filename, 'r') as file: reader = csv.DictReader(file) for row in reader: batch.append(row) if len(batch) >= batch_size: total += process_batch(batch) batch.clear() # Clear batch to free memory  # Process remaining rows  if batch: total += process_batch(batch) return total 

# ✅ DO: Use generators for large sequences data = (x**2 for x in range(1000000)) # ❌ DON'T: Create unnecessary lists data = [x**2 for x in range(1000000)] # ✅ DO: Use context managers with open('file.txt') as f: content = f.read() # ❌ DON'T: Forget to close resources f = open('file.txt') content = f.read() # Forgot to close!  # ✅ DO: Clear references to large objects large_data = process_data() result = extract_result(large_data) del large_data # Free memory  # ❌ DON'T: Keep references unnecessarily cache[key] = large_data # Keeps growing!  # ✅ DO: Use weak references for caches cache = weakref.WeakValueDictionary() # ❌ DON'T: Create circular references without cleanup obj1.ref = obj2 obj2.ref = obj1 # Circular reference!