Python’s weakref module creates references that don’t prevent garbage collection. import weakref. ref: r = weakref.ref(obj) — creates a weak reference; r() returns obj or None if collected. proxy: p = weakref.proxy(obj) — transparent proxy; raises ReferenceError if referent collected. WeakValueDictionary: weakref.WeakValueDictionary() — values are weak refs; entries auto-removed when values are collected. WeakKeyDictionary: weakref.WeakKeyDictionary() — keys are weak refs. WeakSet: weakref.WeakSet() — set with weak element references. finalize: weakref.finalize(obj, callback, *args) — calls callback when obj is collected; .alive property; .cancel() to unregister; .detach() to get obj back. ref callback: weakref.ref(obj, callback) — callback(ref) called on collection. getweakrefcount: weakref.getweakrefcount(obj) — number of weak refs. getweakrefs: weakref.getweakrefs(obj) — list of weak refs. WeakMethod: weakref.WeakMethod(bound_method) — weak ref to a bound method without keeping instance alive. Weakref-able types: user-defined classes, most builtins not weakref-able (int, str, tuple). __weakref__ slot: add __slots__ = ("__weakref__",) to slotted classes. Claude Code generates object caches, observer registries, event buses, and resource lifecycle managers.
CLAUDE.md for weakref
## weakref Stack
- Stdlib: import weakref
- Ref: r = weakref.ref(obj); obj_or_none = r()
- Proxy: p = weakref.proxy(obj) — transparent but raises ReferenceError if dead
- Cache: weakref.WeakValueDictionary() — auto-evicts collected values
- Cleanup: weakref.finalize(obj, fn, *args) — runs fn when obj dies
- Observer: store weakref.ref(callback_obj) not the object itselfweakref Object Lifecycle Pipeline
# app/weakutil.py — weak refs, caches, observer, finalize, WeakMethod
from __future__ import annotations
import gc
import weakref
from collections.abc import Callable
from dataclasses import dataclass, field
from typing import Any, Generic, Iterator, TypeVar
T = TypeVar("T")
K = TypeVar("K")
V = TypeVar("V")
# ─────────────────────────────────────────────────────────────────────────────
# 1. Weak reference helpers
# ─────────────────────────────────────────────────────────────────────────────
def weak_ref(obj: T, callback: Callable | None = None) -> "weakref.ref[T]":
"""
Create a weak reference to obj with an optional dead-notification callback.
Example:
r = weak_ref(some_object, lambda ref: print("collected"))
obj_or_none = r()
"""
if callback is not None:
return weakref.ref(obj, callback)
return weakref.ref(obj)
def deref(ref: "weakref.ref[T]", default: T | None = None) -> T | None:
"""
Dereference a weak ref; return default if the referent has been collected.
Example:
obj = deref(r, default=None)
if obj is None:
print("object is gone")
"""
obj = ref()
return obj if obj is not None else default
def is_alive(ref: "weakref.ref") -> bool:
"""Return True if the referent is still alive."""
return ref() is not None
def safe_proxy(obj: T) -> Any:
"""
Return a weakref proxy to obj.
Accessing attributes through the proxy when obj is collected raises ReferenceError.
Example:
p = safe_proxy(resource)
p.do_work() # transparent; raises ReferenceError if resource collected
"""
return weakref.proxy(obj)
# ─────────────────────────────────────────────────────────────────────────────
# 2. Weak value cache
# ─────────────────────────────────────────────────────────────────────────────
class WeakCache(Generic[K, V]):
"""
A dictionary-like cache whose values are held by weak references.
Entries are automatically removed when their values are garbage collected.
Example:
cache = WeakCache()
cache["img_123"] = load_image("123.png")
# Once no other references to the image exist, it's evicted automatically.
"""
def __init__(self) -> None:
self._store: weakref.WeakValueDictionary[K, V] = weakref.WeakValueDictionary()
def __setitem__(self, key: K, value: V) -> None:
self._store[key] = value
def __getitem__(self, key: K) -> V:
return self._store[key]
def get(self, key: K, default: V | None = None) -> V | None:
return self._store.get(key, default)
def __contains__(self, key: object) -> bool:
return key in self._store
def __delitem__(self, key: K) -> None:
del self._store[key]
def __len__(self) -> int:
return len(self._store)
def keys(self):
return self._store.keys()
def values(self):
return self._store.values()
def items(self):
return self._store.items()
def setdefault(self, key: K, factory: Callable[[], V]) -> V:
"""
Return cached value for key; call factory() and cache it if missing.
Example:
img = cache.setdefault("logo", lambda: load_image("logo.png"))
"""
obj = self._store.get(key)
if obj is None:
obj = factory()
self._store[key] = obj
return obj
def stats(self) -> dict[str, int]:
return {"size": len(self._store)}
# ─────────────────────────────────────────────────────────────────────────────
# 3. Finalize / cleanup callbacks
# ─────────────────────────────────────────────────────────────────────────────
class ResourceGuard:
"""
Attach a cleanup callback to an object that fires when the object is collected.
Works even if the object's __del__ method is not defined.
Example:
conn = open_db_connection()
guard = ResourceGuard(conn, close_db_connection, conn_id)
# When conn is collected, close_db_connection(conn_id) is called.
"""
def __init__(self, obj: Any, cleanup: Callable, *args: Any) -> None:
self._finalizer = weakref.finalize(obj, cleanup, *args)
@property
def alive(self) -> bool:
"""True if the guarded object is still alive."""
return self._finalizer.alive
def cancel(self) -> None:
"""Cancel the cleanup callback (e.g., cleanup already done manually)."""
self._finalizer.cancel()
def __repr__(self) -> str:
return f"ResourceGuard(alive={self.alive})"
def on_collect(obj: T, callback: Callable[[str], None], label: str = "") -> weakref.finalize:
"""
Register a callback that fires with label when obj is garbage collected.
Example:
on_collect(session, lambda lbl: print(f"Session {lbl} ended"), session.id)
"""
return weakref.finalize(obj, callback, label)
# ─────────────────────────────────────────────────────────────────────────────
# 4. Observer / event bus with weak callbacks
# ─────────────────────────────────────────────────────────────────────────────
class WeakObserverList:
"""
A list of weakly-referenced observer callables.
Dead observers are silently dropped during notification.
Example:
obs = WeakObserverList()
class MyListener:
def on_event(self, data):
print("event:", data)
listener = MyListener()
obs.add(listener.on_event)
obs.notify("hello") # prints "event: hello"
del listener
obs.notify("after gc") # silently dropped — no error
"""
def __init__(self) -> None:
self._refs: list[weakref.ref | weakref.WeakMethod] = []
def add(self, callback: Callable) -> None:
"""Register a callback. Uses WeakMethod for bound methods."""
import inspect
if inspect.ismethod(callback):
ref = weakref.WeakMethod(callback)
else:
ref = weakref.ref(callback)
self._refs.append(ref)
def remove(self, callback: Callable) -> None:
"""Unregister a callback (best-effort)."""
self._refs = [r for r in self._refs if r() is not None and r() != callback]
def notify(self, *args: Any, **kwargs: Any) -> int:
"""
Call all live observers with *args/**kwargs.
Returns count of observers notified.
"""
live: list[weakref.ref | weakref.WeakMethod] = []
notified = 0
for ref in self._refs:
cb = ref()
if cb is not None:
live.append(ref)
cb(*args, **kwargs)
notified += 1
self._refs = live
return notified
def __len__(self) -> int:
self._refs = [r for r in self._refs if r() is not None]
return len(self._refs)
class WeakEventBus:
"""
Simple event bus where subscribers are held by weak references.
Example:
bus = WeakEventBus()
class Handler:
def handle(self, payload):
print("got:", payload)
h = Handler()
bus.subscribe("user.created", h.handle)
bus.publish("user.created", {"id": 1})
"""
def __init__(self) -> None:
self._channels: dict[str, WeakObserverList] = {}
def subscribe(self, event: str, callback: Callable) -> None:
if event not in self._channels:
self._channels[event] = WeakObserverList()
self._channels[event].add(callback)
def unsubscribe(self, event: str, callback: Callable) -> None:
if event in self._channels:
self._channels[event].remove(callback)
def publish(self, event: str, *args: Any, **kwargs: Any) -> int:
"""Publish event; return number of live subscribers notified."""
channel = self._channels.get(event)
if channel is None:
return 0
return channel.notify(*args, **kwargs)
def subscriber_counts(self) -> dict[str, int]:
return {ev: len(obs) for ev, obs in self._channels.items()}
# ─────────────────────────────────────────────────────────────────────────────
# 5. WeakKeyDictionary for attribute sidecars
# ─────────────────────────────────────────────────────────────────────────────
class SidecarStore(Generic[K, V]):
"""
Associate extra data with objects without modifying them or keeping them alive.
Uses WeakKeyDictionary: when the object is collected, its sidecar data is too.
Example:
meta = SidecarStore()
meta[request] = {"start_time": time.time(), "user_id": 42}
print(meta.get(request))
del request # sidecar auto-removed
"""
def __init__(self) -> None:
self._store: weakref.WeakKeyDictionary[K, V] = weakref.WeakKeyDictionary()
def __setitem__(self, obj: K, value: V) -> None:
self._store[obj] = value
def __getitem__(self, obj: K) -> V:
return self._store[obj]
def get(self, obj: K, default: V | None = None) -> V | None:
return self._store.get(obj, default)
def __contains__(self, obj: object) -> bool:
return obj in self._store
def __delitem__(self, obj: K) -> None:
del self._store[obj]
def __len__(self) -> int:
return len(self._store)
# ─────────────────────────────────────────────────────────────────────────────
# Demo
# ─────────────────────────────────────────────────────────────────────────────
if __name__ == "__main__":
print("=== weakref demo ===")
print("\n--- basic weak_ref / deref ---")
class Node:
def __init__(self, val: int) -> None:
self.val = val
def __repr__(self) -> str:
return f"Node({self.val})"
n = Node(42)
r = weak_ref(n)
print(f" alive: {is_alive(r)} deref: {deref(r)}")
del n
gc.collect()
print(f" after del: alive={is_alive(r)} deref={deref(r)}")
print("\n--- WeakCache ---")
cache: WeakCache[str, Node] = WeakCache()
a = Node(1)
b = Node(2)
cache["a"] = a
cache["b"] = b
print(f" size: {len(cache)} 'a' in cache: {'a' in cache}")
del a
gc.collect()
print(f" after del a: size={len(cache)} 'a' in cache={'a' in cache}")
print("\n--- ResourceGuard / finalize ---")
collected_labels: list[str] = []
def on_gone(label: str) -> None:
collected_labels.append(label)
resource = Node(99)
guard = ResourceGuard(resource, on_gone, "node-99")
print(f" guard.alive before del: {guard.alive}")
del resource
gc.collect()
print(f" guard.alive after del: {guard.alive}")
print(f" collected_labels: {collected_labels}")
print("\n--- WeakObserverList ---")
results: list[str] = []
class Listener:
def __init__(self, name: str) -> None:
self.name = name
def on_event(self, data: str) -> None:
results.append(f"{self.name}:{data}")
obs = WeakObserverList()
l1 = Listener("L1")
l2 = Listener("L2")
obs.add(l1.on_event)
obs.add(l2.on_event)
obs.notify("ping")
print(f" after notify: {results}")
del l1
gc.collect()
results.clear()
count = obs.notify("ping2")
print(f" after del L1: notified={count} results={results}")
print("\n--- WeakEventBus ---")
bus = WeakEventBus()
payloads: list[Any] = []
class Sub:
def handle(self, data: Any) -> None:
payloads.append(data)
sub = Sub()
bus.subscribe("order.paid", sub.handle)
n1 = bus.publish("order.paid", {"order_id": 1})
print(f" published to {n1} subscriber(s): {payloads}")
del sub
gc.collect()
n2 = bus.publish("order.paid", {"order_id": 2})
print(f" after sub deleted: notified={n2} payloads unchanged={payloads}")
print("\n--- SidecarStore ---")
sidecar: SidecarStore[Node, dict] = SidecarStore()
obj1 = Node(10)
obj2 = Node(20)
sidecar[obj1] = {"tag": "first"}
sidecar[obj2] = {"tag": "second"}
print(f" sidecar[obj1]: {sidecar[obj1]}")
del obj1
gc.collect()
print(f" after del obj1: len={len(sidecar)}")
print("\n=== done ===")For the gc alternative — Python’s gc module (also stdlib) controls the cyclic garbage collector directly: gc.collect() forces a collection cycle, gc.get_referrers(obj) returns all objects holding a reference to obj, gc.is_tracked(obj) checks gc tracking, and gc.callbacks lets you hook collection events; weakref handles reference strength rather than collection timing — use gc when you need to diagnose reference cycles, force immediate reclamation in memory-sensitive hot paths, or hook pre/post-collection events, weakref for designing cache and observer structures where you want Python’s normal collection to remove entries automatically. For the functools.lru_cache alternative — functools.lru_cache (also stdlib) caches the n most recent call results using strong references with a bounded LRU eviction policy; results stay alive until evicted by size pressure; weakref.WeakValueDictionary cache keeps results alive only as long as some other code holds a reference, giving zero size overhead for cold entries — use lru_cache when results should be kept warm between calls and memory pressure is managed by a fixed capacity, WeakValueDictionary cache when objects are large, already held elsewhere by the caller, and you want the cache to act as a free lookup rather than a retention mechanism. The Claude Skills 360 bundle includes weakref skill sets covering weak_ref()/deref()/is_alive()/safe_proxy() reference helpers, WeakCache generic weak-value cache with setdefault() factory, ResourceGuard/on_collect() finalize-based cleanup, WeakObserverList/WeakEventBus observer pattern with automatic dead-subscriber pruning, and SidecarStore WeakKeyDictionary attribute sidecar. Start with the free tier to try object lifecycle management and weakref pipeline code generation.