🎯 Learning Objectives
- Understand CPython's
PyObjectC struct layout and how every Python object is represented in memory - Explain reference counting: how
ob_refcntworks,Py_INCREF/Py_DECREF, and the consequences of cycles - Trace reference count changes using
sys.getrefcount()andctypes - Understand the generational garbage collector: generations, thresholds, and the
gcmodule - Identify and break reference cycles with
__del__,gc.collect(), andweakref - Use
weakref.ref,WeakValueDictionary, andWeakSetfor cache and observer patterns - Understand object interning, small-integer caching, and string interning — and when
islies
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.hThe 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_deallocis called and memory is freed immediately.ob_type: pointer to the type object.type(x)in Python is justx->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.pyid(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 xis 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.pyReference 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 collectcycle_leak.py__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:
| Generation | Contains | Default Threshold | Collection Frequency |
|---|---|---|---|
| 0 | Newly allocated objects | 700 allocations | Most frequent |
| 1 | Survived one gen-0 collection | 10 gen-0 collections | Less often |
| 2 | Long-lived objects | 10 gen-1 collections | Rarely |
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.pyComplete 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.pygc.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 runcontext_manager.pyatexit, 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 Trueinterning_demo.pyThe rules:
- Small ints (
-5to256): 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.
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 integerssizeof_basics.pyTo 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 integersdeep_sizeof.pyPython's string internals use three encodings (PEP 393 — flexible string representation):
| Encoding | Bytes per Char | Used When |
|---|---|---|
| Latin-1 | 1 | All code points ≤ U+00FF |
| UCS-2 | 2 | Highest code point ≤ U+FFFF |
| UCS-4 | 4 | Any 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.pyThe 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-unregisteredevent_bus_weakset.pyWeakSet 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
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
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
isfor value comparison — only useisforNone,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. Usewith/__enter__/__exit__for deterministic cleanup. - Break large cycles explicitly — if you must have a cycle (e.g. parent ↔ child), use a
weakref.reffor the back-pointer to avoid keeping both objects alive. - Use
WeakValueDictionaryfor caches — entries are automatically evicted when the cached value is no longer referenced elsewhere, preventing unbounded growth. - Always define
__hash__when you define__eq__— usehash(tuple_of_fields)for immutable classes; set__hash__ = Noneexplicitly for mutable classes that should be unhashable. - Use
tracemallocto 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__, recordsys.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.WeakSetto store subscribers per event name. - Support
subscribe(event, handler),unsubscribe(event, handler), andpublish(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)