🟠 Python Internals

The Python Object Model: Memory, gc & Weak References

📖 Lesson 44 ⏱ 60 min 🧪 5 questions 💻 3 exercises

🎯 Learning Objectives

  • Understand CPython's PyObject C struct layout and how every Python object is represented in memory
  • Explain reference counting: how ob_refcnt works, Py_INCREF/Py_DECREF, and the consequences of cycles
  • Trace reference count changes using sys.getrefcount() and ctypes
  • Understand the generational garbage collector: generations, thresholds, and the gc module
  • Identify and break reference cycles with __del__, gc.collect(), and weakref
  • Use weakref.ref, WeakValueDictionary, and WeakSet for cache and observer patterns
  • Understand object interning, small-integer caching, and string interning — and when is lies

1 — Every Object is a PyObject

In CPython, every Python object — integer, string, list, class instance — is a C struct that starts with PyObject_HEAD. This is the fundamental building block of CPython's object system: a uniform header that gives the runtime everything it needs to manage memory and dispatch type operations.

/* Every Python object starts with this header */
typedef struct _object {
    Py_ssize_t ob_refcnt;    /* reference count — atomic in free-threaded mode */
    PyTypeObject *ob_type;   /* pointer to the type object (int, str, list …) */
} PyObject;

/* Objects with variable length (list, tuple, str) add ob_size */
typedef struct {
    PyObject_HEAD
    Py_ssize_t ob_size;      /* number of items */
} PyVarObject;

/* Example: a Python integer */
typedef struct {
    PyObject_HEAD
    _PyLongValue long_value; /* actual integer value (arbitrary precision) */
} PyLongObject;
CPython/Objects/object.h

The key fields:

  • ob_refcnt: the reference count. Incremented every time a new reference to the object is created; decremented when a reference is removed. When it hits zero, tp_dealloc is called and memory is freed immediately.
  • ob_type: pointer to the type object. type(x) in Python is just x->ob_type->tp_name. This is how Python's type system works — every object knows its type via this single pointer.
  • Memory layout is compact and cache-friendly for simple types; for Python-level objects __dict__ is a separate heap allocation.
import sys
import ctypes

x = [1, 2, 3]

# How many bytes does this object occupy?
print(sys.getsizeof(x))           # 88 bytes (list shell; elements not counted)
print(sys.getsizeof(x[0]))        # 28 bytes (small int — CPython 3.12)

# Reference count (getrefcount adds 1 for the argument itself)
print(sys.getrefcount(x))         # 2 (x + getrefcount's argument)

# Read ob_refcnt directly via ctypes
id_x = id(x)
refcnt = ctypes.c_ssize_t.from_address(id_x).value
print(f"ob_refcnt via ctypes: {refcnt}")

# ob_type pointer
ob_type_ptr = ctypes.c_size_t.from_address(id_x + ctypes.sizeof(ctypes.c_ssize_t)).value
print(f"ob_type pointer: 0x{ob_type_ptr:x}")
print(f"type(x): {type(x)}")
pyobject_inspect.py
Concept: id(obj) in CPython returns the memory address of the object — its PyObject* pointer. This is CPython-specific; PyPy, Jython, and other implementations have different id() semantics.

2 — Reference Counting in Depth

Reference counting is CPython's primary memory management mechanism. A reference count is incremented when:

  • A name binding is created: x = obj
  • An object is added to a container: lst.append(obj)
  • An object is passed as a function argument
  • An object is returned from a function
  • An attribute is set: self.attr = obj

And decremented when:

  • A name goes out of scope
  • del x is executed
  • A container is cleared or destroyed
  • A function call returns
import sys

a = []             # refcount = 1  (a)
b = a              # refcount = 2  (a, b)
c = [a]            # refcount = 3  (a, b, c[0])

print(sys.getrefcount(a))  # 4 — +1 for getrefcount's own argument

del b              # refcount = 3
c.clear()          # refcount = 2 (a, getrefcount arg)

# When refcount hits 0, __del__ is called immediately (if defined)
class Tracked:
    def __init__(self, name): self.name = name
    def __del__(self):        print(f"{self.name} freed")

t = Tracked("A")   # refcount=1
del t              # prints "A freed" immediately — deterministic!
refcount_demo.py

Reference counting alone cannot handle cycles:

# Cycle: a → b → a
a = {}
b = {"ref": a}
a["ref"] = b

import sys
print(sys.getrefcount(a))   # 3 (a, b["ref"], getrefcount arg)
print(sys.getrefcount(b))   # 3

del a
del b
# Both still have refcount=1 from each other — never freed by refcounting alone
# Requires the cyclic GC to collect
cycle_leak.py
Warning: Objects with __del__ methods in a reference cycle used to be uncollectable (Python ≤ 3.3) because CPython couldn't determine safe finalisation order. Python 3.4+ (PEP 442) fixed this — __del__ is called even for objects in cycles, but the order is still undefined. Avoid relying on __del__ for critical cleanup; use context managers instead.

3 — The Generational Garbage Collector

CPython's cyclic GC (gc module) complements reference counting by collecting objects involved in reference cycles. It uses a generational scheme based on the empirical observation that most objects are short-lived.

Three generations based on object age and survival:

GenerationContainsDefault ThresholdCollection Frequency
0Newly allocated objects700 allocationsMost frequent
1Survived one gen-0 collection10 gen-0 collectionsLess often
2Long-lived objects10 gen-1 collectionsRarely
import gc

# Current thresholds
print(gc.get_threshold())     # (700, 10, 10) by default

# Force a full collection
collected = gc.collect()
print(f"Collected {collected} objects")

# Collect only generation 0
gc.collect(0)

# Disable the cyclic GC (useful in scripts with no cycles, saves ~10% overhead)
gc.disable()
# ... do work ...
gc.enable()

# Manual generation inspection
print(gc.get_count())         # (n0, n1, n2) — current object counts per gen

# Find what objects are tracked
tracked = gc.get_objects(generation=0)
print(f"Gen 0 tracked: {len(tracked)} objects")

# Diagnostics: what's in a cycle?
gc.set_debug(gc.DEBUG_SAVEALL)  # save unreachable objects to gc.garbage
gc.collect()
if gc.garbage:
    print("Uncollected:", gc.garbage)
gc.set_debug(0)
gc_exploration.py

Complete cycle creation → detection → collection example:

import gc

class Node:
    def __init__(self, name):
        self.name = name
        self.partner = None
    def __repr__(self):
        return f"Node({self.name!r})"
    def __del__(self):
        print(f"  🗑️  {self.name} collected")

# Disable automatic GC so we control timing
gc.disable()

# Create a cycle
n1 = Node("alpha")
n2 = Node("beta")
n1.partner = n2
n2.partner = n1

print("Before del — objects alive")
del n1
del n2
print("After del — objects STILL alive (cycle holds them)")

# Now trigger collection
print("Running gc.collect()...")
collected = gc.collect()
print(f"Collected {collected} objects")  # Prints "alpha collected", "beta collected"

gc.enable()
gc_cycle_demo.py
Tip: In long-running server processes you can tune GC thresholds: gc.set_threshold(1000, 15, 15) reduces GC frequency at the cost of higher peak memory. Facebook's Instagram team famously disabled gen-2 GC entirely for their Django workers and saw ~10% throughput improvement because their requests were short-lived and cycle-free.

4 — __del__, Finalisation & Context Managers

import gc, weakref

class Resource:
    _count = 0

    def __init__(self, name: str):
        self.name = name
        Resource._count += 1
        print(f"  [+] {name} created  (total={Resource._count})")

    def __del__(self):
        Resource._count -= 1
        print(f"  [-] {self.name} freed  (total={Resource._count})")

# Normal case: freed deterministically when refcount → 0
r = Resource("R1")
del r      # prints immediately

# Cycle case: NOT freed until GC runs
r2 = Resource("R2")
r2.self_ref = r2   # cycle!
del r2
print("After del r2 (still alive due to cycle)")
gc.collect()       # now freed
print("After gc.collect()")
finalisation.py
# The __del__ gotcha: accessing global state during interpreter shutdown
# is dangerous — globals may already be None
import atexit

class SafeResource:
    def __init__(self, name):
        self.name = name

    def close(self):
        print(f"Closing {self.name}")

    def __enter__(self):
        return self

    def __exit__(self, *exc):
        self.close()
        return False

# Always prefer context managers over __del__ for deterministic cleanup
with SafeResource("DB connection") as res:
    pass   # __exit__ guaranteed to run
context_manager.py
Warning: During interpreter shutdown (atexit, module teardown), global variables are set to None in an unspecified order. If __del__ tries to access a module-level name (e.g. open, print) it may get None and raise TypeError. This is why CPython sometimes prints Exception ignored in: <function ...> during shutdown.

5 — Object Interning & Identity

CPython interns (reuses) certain objects to save memory and speed up comparisons:

# ── Small integers: cached in range [-5, 256] ──
a = 256;  b = 256;  print(a is b)    # True  — same object
a = 257;  b = 257;  print(a is b)    # False — different objects (CPython)

# ── String interning ──
s1 = "hello"
s2 = "hello"
print(s1 is s2)        # True  — interned (looks like an identifier)

s3 = "hello world"
s4 = "hello world"
print(s3 is s4)        # True in CPython (compile-time constant folding)
# But:
s5 = "hello" + " world"   # runtime concatenation
s6 = "hello" + " world"
print(s5 is s6)            # False — NOT the same object

import sys
s7 = sys.intern("my-non-identifier string!")
s8 = sys.intern("my-non-identifier string!")
print(s7 is s8)            # True — explicitly interned

# ── Tuple interning ──
t1 = ()
t2 = ()
print(t1 is t2)   # True — empty tuple is singleton

t3 = (1, 2)
t4 = (1, 2)
print(t3 is t4)   # True in CPython (small constant tuples may be interned)

# ── None, True, False are singletons ──
print(None is None)    # always True
print(True  is True)   # always True
interning_demo.py

The rules:

  • Small ints (-5 to 256): cached at interpreter startup in a pre-allocated array.
  • Compile-time string constants that look like identifiers: interned automatically.
  • sys.intern(): explicitly intern any string — useful for dictionary keys that are looked up millions of times.
  • Empty tuples and None/True/False: singletons.
Warning: Never use is to compare values — use ==. is tests identity (same memory address), not equality. a is b means id(a) == id(b), which is only meaningful for singletons and interned objects. Writing if x is "active": is a bug waiting to happen — use if x == "active":.

6 — sys.getsizeof & Memory Layout

import sys

# Base sizes (CPython 3.12, 64-bit Linux)
print(sys.getsizeof(None))           # 16
print(sys.getsizeof(True))           # 28
print(sys.getsizeof(0))              # 28
print(sys.getsizeof(2**30))          # 32  (larger int, more limbs)
print(sys.getsizeof(2**300))         # 68
print(sys.getsizeof(""))             # 49
print(sys.getsizeof("a"))            # 50  (+1 per Latin-1 char)
print(sys.getsizeof("α"))            # 76  (UCS-2 — non-Latin chars)
print(sys.getsizeof([]))             # 56  (empty list shell)
print(sys.getsizeof([1]))            # 64  (+8 bytes per pointer)
print(sys.getsizeof({}))             # 64  (empty dict)
print(sys.getsizeof(set()))          # 216 (empty set — pre-allocated hash table)
print(sys.getsizeof(lambda: None))   # 144

# getsizeof does NOT include referenced objects
lst = [1, 2, 3]
print(sys.getsizeof(lst))            # 88 — shell only, not the integers
sizeof_basics.py

To get the deep (recursive) size of an object graph:

import sys, gc

def deep_sizeof(obj, seen=None):
    if seen is None: seen = set()
    obj_id = id(obj)
    if obj_id in seen: return 0
    seen.add(obj_id)
    size = sys.getsizeof(obj)
    if isinstance(obj, dict):
        size += sum(deep_sizeof(k, seen) + deep_sizeof(v, seen) for k, v in obj.items())
    elif hasattr(obj, '__dict__'):
        size += deep_sizeof(obj.__dict__, seen)
    elif hasattr(obj, '__iter__') and not isinstance(obj, (str, bytes)):
        size += sum(deep_sizeof(i, seen) for i in obj)
    return size

print(deep_sizeof([1, 2, 3]))   # ~140 bytes including the integers
deep_sizeof.py

Python's string internals use three encodings (PEP 393 — flexible string representation):

EncodingBytes per CharUsed When
Latin-11All code points ≤ U+00FF
UCS-22Highest code point ≤ U+FFFF
UCS-44Any code point > U+FFFF (emoji, rare CJK)

The encoding is chosen based on the highest code point in the string. A single emoji in an otherwise ASCII string forces the entire string to UCS-4 (4 bytes per character).

7 — Weak References

A weak reference does not increment the reference count — the referenced object can be garbage-collected even while the weak reference exists. This is critical for caches, observer patterns, and any structure where you want to observe an object without owning it.

import weakref
import gc

class BigCache:
    def __init__(self, key): self.key = key
    def __repr__(self): return f"BigCache({self.key!r})"

# ── Basic weak reference ──
obj = BigCache("data")
ref = weakref.ref(obj)

print(ref())        #   — object alive
print(ref() is obj) # True

del obj
gc.collect()
print(ref())        # None — object was collected

# ── Callback on collection ──
def on_finalize(ref):
    print(f"Object collected: {ref}")

obj2 = BigCache("x")
ref2 = weakref.ref(obj2, on_finalize)
del obj2   # prints "Object collected: "

# ── WeakValueDictionary — auto-evicts dead entries ──
cache: weakref.WeakValueDictionary[str, BigCache] = weakref.WeakValueDictionary()
item = BigCache("item1")
cache["item1"] = item

print(dict(cache))   # {'item1': BigCache('item1')}
del item
gc.collect()
print(dict(cache))   # {}  — entry automatically removed

# ── WeakSet — set of weakly-referenced objects ──
class Subscriber:
    def __init__(self, name): self.name = name

ws: weakref.WeakSet[Subscriber] = weakref.WeakSet()
s1 = Subscriber("Alice")
s2 = Subscriber("Bob")
ws.add(s1); ws.add(s2)
print([s.name for s in ws])   # ['Alice', 'Bob']
del s1
gc.collect()
print([s.name for s in ws])   # ['Bob']  — Alice auto-removed
weakref_basics.py

The observer/event-bus pattern using WeakSet — subscribers that go out of scope are automatically unregistered, preventing memory leaks:

import weakref

class EventBus:
    def __init__(self):
        self._handlers: weakref.WeakSet = weakref.WeakSet()

    def subscribe(self, handler):
        self._handlers.add(handler)

    def publish(self, event):
        for h in list(self._handlers):
            h(event)

bus = EventBus()

class Handler:
    def __init__(self, name): self.name = name
    def __call__(self, event): print(f"{self.name} received: {event}")

h1 = Handler("Logger")
h2 = Handler("Alerter")
bus.subscribe(h1); bus.subscribe(h2)

bus.publish("login")   # both fire
del h1
import gc; gc.collect()
bus.publish("logout")  # only Alerter fires — Logger auto-unregistered
event_bus_weakset.py
Concept: WeakSet requires objects to be hashable and weak-referenceable. Most user-defined classes are both by default. Types that do NOT support weak references: int, str, tuple, bytes, bool — because they lack the tp_weaklistoffset slot in their C struct. Add __weakref__ to your class (or inherit from a class that has it) if you're using __slots__.

Memory Pools & the Object Allocator

CPython does not call malloc() / free() for every object. It uses a three-tier allocator to avoid fragmentation and allocation overhead.

Tier 3 — OS / malloc (objects > 512 bytes)
    ↑ falls through for large allocations

Tier 2 — pymalloc (objects ≤ 512 bytes)
    Manages Arenas (256 KB blocks from the OS)
    Each Arena is divided into Pools (4 KB each)
    Each Pool holds fixed-size Blocks for one size class

Tier 1 — Object-specific allocators
    e.g. intobject freelist (small ints cached), listobject freelist,
    frameobject freelist — recycle recently freed objects of the same type
    

The pymalloc arena/pool/block hierarchy means that allocating a 20-byte Python dict entry does not involve a kernel call — it just pops the next free block from the appropriate pool's free-list.

import sys

# Object freelists — CPython recycles recently freed objects
# E.g. list freelist: up to 80 empty list shells are kept ready

a = []
id_a = id(a)
del a
b = []
print(id(b) == id_a)   # True on CPython — same memory address reused from freelist!

# Frame freelist: function call overhead is low partly because
# frame objects are recycled from a per-type freelist
import dis

def inner(): pass
def outer():
    for _ in range(5):
        inner()   # frame recycled each iteration

# tracemalloc: track Python-level allocations
import tracemalloc
tracemalloc.start()

data = [dict(x=i) for i in range(10_000)]

snapshot = tracemalloc.take_snapshot()
stats = snapshot.statistics("lineno")
for stat in stats[:5]:
    print(stat)
allocator.py
Use tracemalloc to hunt memory leaks in long-running services. tracemalloc.take_snapshot() records per-line allocation statistics. Compare two snapshots with snapshot2.compare_to(snapshot1, "lineno") to see which lines are allocating the most memory between two points in time.

Detecting & Fixing Memory Leaks

In Python, a "memory leak" usually means objects are being kept alive unintentionally — by a reference cycle, a global cache that never evicts, or a closure holding a large object.

import gc, tracemalloc, weakref

# ── Pattern 1: Unbounded global cache ──
_cache: dict = {}

def get_data(key):
    if key not in _cache:
        _cache[key] = {"data": b"x" * 1024 * 10}   # 10 KB per entry
    return _cache[key]

# Fix: use WeakValueDictionary or limit size with functools.lru_cache
from functools import lru_cache

@lru_cache(maxsize=128)
def get_data_cached(key):
    return {"data": b"x" * 1024 * 10}   # auto-evicts oldest entries

# ── Pattern 2: Cycle with __del__ ──
class Node:
    def __init__(self, val):
        self.val = val
        self.next = None
    def __del__(self):
        pass   # even trivial __del__ can trap cycles pre-3.4

a = Node(1)
b = Node(2)
a.next = b
b.next = a    # cycle
del a, b
print(f"Collected: {gc.collect()}")   # 2

# ── Pattern 3: Closures holding large objects ──
def make_leak():
    large = list(range(100_000))   # 800 KB
    def inner():
        return large[0]            # inner holds ref to large forever
    return inner                   # large lives as long as inner lives

# Fix: only capture what you need
def make_fixed():
    large = list(range(100_000))
    first = large[0]               # capture only the scalar
    del large                      # large freed here
    def inner():
        return first
    return inner

# ── tracemalloc snapshot diff ──
tracemalloc.start()
snap1 = tracemalloc.take_snapshot()

data = [Node(i) for i in range(1000)]

snap2 = tracemalloc.take_snapshot()
for stat in snap2.compare_to(snap1, "lineno")[:3]:
    print(stat)
leaks.py

Identity, Equality & the Hash Protocol

Understanding the relationship between is, ==, id(), __hash__, and __eq__ is essential for writing correct Python — and for understanding how dicts and sets work internally.

# ── Identity vs Equality ──
a = [1, 2, 3]
b = [1, 2, 3]
c = a

print(a == b)      # True  — same value (__eq__)
print(a is b)      # False — different objects (different id)
print(a is c)      # True  — same object

# ── Hash contract: equal objects MUST have equal hashes ──
# If __eq__ is defined, __hash__ must also be defined (or set to None for unhashable)

class Point:
    def __init__(self, x, y):
        self.x, self.y = x, y

    def __eq__(self, other):
        return isinstance(other, Point) and self.x == other.x and self.y == other.y

    def __hash__(self):
        return hash((self.x, self.y))   # tuple hash — fast, well-distributed

p1 = Point(1, 2)
p2 = Point(1, 2)
print(p1 == p2)    # True
print(p1 is p2)    # False
print(hash(p1) == hash(p2))  # True
print({p1, p2})    # {Point(1,2)} — deduplicated in set

# ── Hash collisions — still equal objects must compare equal ──
# Python's dict/set resolves collisions by calling __eq__ after hash match
# hash("abc") == hash("abc") — always
# Two distinct objects CAN have the same hash (collision), so __eq__ is the tiebreaker

# ── Mutable objects and hashing ──
lst = [1, 2, 3]
# hash(lst)  → TypeError: unhashable type: 'list'
# Lists define __eq__ but set __hash__ = None → unhashable

# ── id() uniqueness guarantee ──
# id() is unique only for SIMULTANEOUSLY ALIVE objects
# After del x, a new object can get the same id
x = object()
xid = id(x)
del x
y = object()
print(id(y) == xid)   # Possibly True! ids can be reused after deallocation

# ── Interning and the is trap ──
import sys
a = sys.intern("cached_key")
b = sys.intern("cached_key")
print(a is b)   # True  — safe to use is for interned strings in caches

# ── __eq__ without __hash__ ──
class NoHash:
    def __eq__(self, other): return True
    # __hash__ automatically set to None by Python when __eq__ is defined without __hash__

nh = NoHash()
try:
    hash(nh)
except TypeError as e:
    print(e)   # unhashable type: 'NoHash'
identity_equality.py
The hash contract: if a == b, then hash(a) == hash(b) MUST hold. The converse is not required (hash collisions are allowed). Python enforces this by setting __hash__ = None whenever you define __eq__ without also defining __hash__ — making the object unhashable and preventing it from being used as a dict key or set member.

Best Practices

  • Never use is for value comparison — only use is for None, True, False, and explicitly interned singletons. All other comparisons must use ==.
  • Prefer context managers over __del____del__ is non-deterministic in cycles, dangerous during shutdown, and easy to misuse. Use with / __enter__ / __exit__ for deterministic cleanup.
  • Break large cycles explicitly — if you must have a cycle (e.g. parent ↔ child), use a weakref.ref for the back-pointer to avoid keeping both objects alive.
  • Use WeakValueDictionary for caches — entries are automatically evicted when the cached value is no longer referenced elsewhere, preventing unbounded growth.
  • Always define __hash__ when you define __eq__ — use hash(tuple_of_fields) for immutable classes; set __hash__ = None explicitly for mutable classes that should be unhashable.
  • Use tracemalloc to profile allocations — compare snapshots before and after suspected leaky operations to pinpoint the source file and line.
  • Add __weakref__ to __slots__ classes — without it, instances of a __slots__ class cannot be weak-referenced: __slots__ = ('x', 'y', '__weakref__').
  • Tune GC thresholds for server workloads — short-lived request-handling processes with few cycles can safely increase gen-0 threshold or disable gen-2 collection to reduce GC pauses.

Exercises

Exercise 1 — Reference Count Tracker

Build a context manager RefCountMonitor that tracks the reference count delta of a specific object across the block:

  • On __enter__, record sys.getrefcount(obj) - 1 (subtract the monitor's own reference).
  • On __exit__, record it again and print the delta.
  • Test it across: simple name assignment, appending to a list, passing as an argument, storing as an attribute, and deleting the reference.
  • Verify that passing to a function adds +1 during the call and returns to baseline after.
💡 Hint
import sys

class RefCountMonitor:
    def __init__(self, obj, label=""):
        self.obj   = obj
        self.label = label

    def count(self):
        # subtract 1 for self.obj, 1 for getrefcount arg
        return sys.getrefcount(self.obj) - 2

    def __enter__(self):
        self._start = self.count()
        print(f"[{self.label}] start refcount = {self._start}")
        return self

    def __exit__(self, *_):
        end = self.count()
        print(f"[{self.label}] end refcount = {end}  (delta={end - self._start})")

x = object()
with RefCountMonitor(x, "basic") as m:
    y = x             # +1
    lst = [x, x]      # +2
print()
# y and lst go out of scope at the end of the with block

Exercise 2 — Cycle Detector

Write a function find_cycles(obj) that detects whether a given object is part of a reference cycle without using the gc module's built-in cycle finder:

  • Use gc.get_referents(obj) to walk the object graph.
  • Track visited object IDs to detect when you encounter an already-seen object.
  • Return a list of (object, referrer_chain) tuples for each cycle found.
  • Test with: a simple cycle (a.x = a), a two-node cycle (a.x = b; b.x = a), and a cycle-free object graph.
  • Compare your results against gc.collect() output.
💡 Hint
import gc

def find_cycles(root, _seen=None, _path=None):
    if _seen is None: _seen = {}
    if _path is None: _path = []

    obj_id = id(root)
    if obj_id in _seen:
        return [_path + [root]]   # cycle detected!

    _seen = dict(_seen)
    _seen[obj_id] = root
    _path = _path + [root]

    cycles = []
    for ref in gc.get_referents(root):
        if not isinstance(ref, type):   # skip type objects
            cycles.extend(find_cycles(ref, _seen, _path))
    return cycles

# Test
class Node:
    def __init__(self, v): self.v, self.next = v, None

a = Node(1); b = Node(2)
a.next = b; b.next = a   # cycle
print(f"Cycles found: {len(find_cycles(a))}")

Exercise 3 — Weak-Reference Event Bus

Build a production-quality EventBus using weak references:

  • Use weakref.WeakSet to store subscribers per event name.
  • Support subscribe(event, handler), unsubscribe(event, handler), and publish(event, *args, **kwargs).
  • When a subscriber object is garbage-collected, it should automatically be removed from all subscriptions without any explicit unsubscribe call.
  • Write tests: (a) subscriber receives events while alive; (b) after del subscriber + gc.collect(), no dead handlers are called and no errors are raised; (c) explicitly unsubscribing works.
  • Add a subscriber_count(event) method that returns the number of live subscribers.
💡 Hint
import weakref, gc
from collections import defaultdict

class EventBus:
    def __init__(self):
        self._subs: dict[str, weakref.WeakSet] = defaultdict(weakref.WeakSet)

    def subscribe(self, event: str, handler) -> None:
        self._subs[event].add(handler)

    def unsubscribe(self, event: str, handler) -> None:
        self._subs[event].discard(handler)

    def publish(self, event: str, *args, **kwargs) -> int:
        called = 0
        for h in list(self._subs.get(event, [])):
            h(*args, **kwargs)
            called += 1
        return called

    def subscriber_count(self, event: str) -> int:
        return len(self._subs.get(event, []))

# Test
bus = EventBus()
log = []

class Handler:
    def __init__(self, name): self.name = name
    def __call__(self, msg):  log.append(f"{self.name}: {msg}")

h1, h2 = Handler("A"), Handler("B")
bus.subscribe("msg", h1)
bus.subscribe("msg", h2)
bus.publish("msg", "hello")     # both fire
del h1
gc.collect()
bus.publish("msg", "world")     # only B fires
print(log)