Python weakref: Avoiding Memory Leaks with Weak References

import weakref
 
class ExpensiveObject:
    def __init__(self, name):
        self.name = name
    
    def __del__(self):
        print(f"{self.name} deleted")
 
# Strong reference
obj = ExpensiveObject("test")
 
# Create weak reference
weak = weakref.ref(obj)
 
# Access through weak reference
print(weak())  # <ExpensiveObject object>
print(weak().name)  # "test"
 
# Delete strong reference
del obj
 
# Weak reference now returns None
print(weak())  # None

import weakref
 
def callback(ref):
    print(f"Object was garbage collected!")
 
class Resource:
    pass
 
obj = Resource()
weak = weakref.ref(obj, callback)
 
del obj
# Output: "Object was garbage collected!"

import weakref
 
class User:
    def __init__(self, user_id, name):
        self.user_id = user_id
        self.name = name
 
# Cache users without preventing cleanup
user_cache = weakref.WeakValueDictionary()
 
def get_user(user_id):
    if user_id in user_cache:
        return user_cache[user_id]
    
    # Fetch from database
    user = User(user_id, f"User {user_id}")
    user_cache[user_id] = user
    return user
 
# Usage
user = get_user(1)
print(1 in user_cache)  # True
 
del user
# Entry automatically removed when user is garbage collected

import weakref
 
class Widget:
    def __init__(self, name):
        self.name = name
 
# Track metadata without preventing widget deletion
metadata = weakref.WeakKeyDictionary()
 
widget = Widget("button")
metadata[widget] = {"created": "2024-01-01", "clicks": 0}
 
print(metadata[widget])  # {'created': '2024-01-01', 'clicks': 0}
 
del widget
# Entry automatically removed

import weakref
 
class Observer:
    def __init__(self, name):
        self.name = name
    
    def notify(self, message):
        print(f"{self.name} received: {message}")
 
class Subject:
    def __init__(self):
        self._observers = weakref.WeakSet()
    
    def attach(self, observer):
        self._observers.add(observer)
    
    def notify_all(self, message):
        for observer in self._observers:
            observer.notify(message)
 
# Usage
subject = Subject()
 
obs1 = Observer("Observer1")
obs2 = Observer("Observer2")
 
subject.attach(obs1)
subject.attach(obs2)
 
subject.notify_all("Hello!")  # Both notified
 
del obs1  # Automatically removed from observers
subject.notify_all("Goodbye!")  # Only obs2 notified

import weakref
 
# Problem: circular reference prevents garbage collection
class Parent:
    def __init__(self, name):
        self.name = name
        self.children = []
 
class ChildBad:
    def __init__(self, parent):
        self.parent = parent  # Strong reference creates cycle!
 
# Solution: weak reference to parent
class ChildGood:
    def __init__(self, parent):
        self._parent_ref = weakref.ref(parent)
    
    @property
    def parent(self):
        return self._parent_ref()
 
# Usage
parent = Parent("parent")
child = ChildGood(parent)
parent.children.append(child)
 
print(child.parent.name)  # "parent"
 
del parent
print(child.parent)  # None (parent was collected)

import weakref
 
class TempFile:
    def __init__(self, path):
        self.path = path
        print(f"Created temp file: {path}")
        
        # Register finalizer
        self._finalizer = weakref.finalize(
            self, 
            self._cleanup, 
            path
        )
    
    @staticmethod
    def _cleanup(path):
        print(f"Cleaning up: {path}")
        # os.unlink(path)  # Would delete file
    
    def close(self):
        """Manually trigger cleanup."""
        self._finalizer()
 
# Automatic cleanup
temp = TempFile("/tmp/test.txt")
del temp  # Output: "Cleaning up: /tmp/test.txt"
 
# Or manual cleanup
temp = TempFile("/tmp/test2.txt")
temp.close()  # Output: "Cleaning up: /tmp/test2.txt"
del temp  # No output (already cleaned)

import weakref
from collections import OrderedDict
 
class WeakLRUCache:
    """LRU cache that uses weak references for values."""
    
    def __init__(self, maxsize=128):
        self.maxsize = maxsize
        self.cache = OrderedDict()
        self._refs = weakref.WeakValueDictionary()
    
    def get(self, key):
        if key in self._refs:
            value = self._refs[key]
            if value is not None:
                # Move to end (most recently used)
                self.cache.move_to_end(key)
                return value
        return None
    
    def set(self, key, value):
        if key in self.cache:
            self.cache.move_to_end(key)
        else:
            if len(self.cache) >= self.maxsize:
                self.cache.popitem(last=False)
        
        self.cache[key] = value
        self._refs[key] = value

import weakref
 
class ExpensiveResource:
    def __init__(self, data):
        self.data = data
    
    def process(self):
        return f"Processed: {self.data}"
 
# Create resource
resource = ExpensiveResource("important data")
 
# Create proxy (acts like the object)
proxy = weakref.proxy(resource)
 
# Use proxy like the original
print(proxy.data)       # "important data"
print(proxy.process())  # "Processed: important data"
 
# After deletion, proxy raises ReferenceError
del resource
try:
    print(proxy.data)
except ReferenceError:
    print("Object no longer exists")

import weakref
 
class TrackedObject:
    _instances = weakref.WeakSet()
    
    def __init__(self, name):
        self.name = name
        TrackedObject._instances.add(self)
    
    @classmethod
    def get_all_instances(cls):
        return list(cls._instances)
 
# Create objects
obj1 = TrackedObject("first")
obj2 = TrackedObject("second")
obj3 = TrackedObject("third")
 
print([o.name for o in TrackedObject.get_all_instances()])
# ['first', 'second', 'third']
 
del obj2
print([o.name for o in TrackedObject.get_all_instances()])
# ['first', 'third']

import weakref
 
# These raise TypeError
# weakref.ref(42)
# weakref.ref("hello")
# weakref.ref((1, 2, 3))
 
# These work
weakref.ref([1, 2, 3])     # list
weakref.ref({1: 2})        # dict
weakref.ref(lambda x: x)   # function
weakref.ref(object())      # custom objects