feat: add EventBus.remove() to unsubscribe handlers

bubus/service.py

@@ -453,17 +453,7 @@ def on(
             f'Invalid handler: {handler}, must be a sync or async function or method'
         )
 
-        # Determine event key
-        event_key: str
-        if event_pattern == '*':
-            event_key = '*'
-        elif isinstance(event_pattern, type) and issubclass(event_pattern, BaseEvent):  # pyright: ignore[reportUnnecessaryIsInstance]
-            event_key = event_pattern.__name__  # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]
-        else:
-            event_key = str(event_pattern)
-
-        # Ensure event_key is definitely a string at this point
-        assert isinstance(event_key, str)
+        event_key = self._get_event_key(event_pattern)
 
         # Check for duplicate handler names
         new_handler_name = get_handler_name(handler)
@@ -482,6 +472,38 @@ def on(
         self.handlers[event_key].append(handler)  # type: ignore
         logger.debug(f'👂 {self}.on({event_key}, {get_handler_name(handler)}) Registered event handler')
 
+    def _get_event_key(self, event_pattern: EventPatternType) -> str:
+        """Resolve an event pattern (type name string, event class, or '*') to the internal handler key."""
+        if event_pattern == '*':
+            return '*'
+        elif isinstance(event_pattern, type) and issubclass(event_pattern, BaseEvent):  # pyright: ignore[reportUnnecessaryIsInstance]
+            return event_pattern.__name__  # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]
+        return str(event_pattern)
+
+    def remove(self, event_pattern: EventPatternType, handler: ContravariantEventHandler['BaseEvent[Any]']) -> bool:
+        """
+        Unsubscribe a previously registered handler from events matching a pattern.
+
+        Args:
+                event_pattern: The event type string, event model class, or '*' that was passed to `on()`.
+                handler: The exact handler function or method previously registered with `on()`.
+
+        Returns:
+                True if the handler was found and removed, False otherwise.
+
+        Example:
+                eventbus.on(TaskStartedEvent, handler)
+                ...
+                eventbus.remove(TaskStartedEvent, handler)
+        """
+        event_key = self._get_event_key(event_pattern)
+        registered_handlers = self.handlers.get(event_key, [])
+        if handler in registered_handlers:
+            registered_handlers.remove(handler)
+            logger.debug(f'👂 {self}.remove({event_key}, {get_handler_name(handler)}) Unregistered event handler')
+            return True
+        return False
+
     def dispatch(self, event: T_ExpectedEvent) -> T_ExpectedEvent:
         """
         Enqueue an event for processing and immediately return an Event(status='pending') version (synchronous).
@@ -677,9 +699,7 @@ def notify_expect_handler(event: 'BaseEvent[Any]') -> None:
                 return await future
         finally:
             # Clean up handler
-            event_key: str = event_type.__name__ if isinstance(event_type, type) else str(event_type)  # pyright: ignore[reportUnknownMemberType, reportPartialTypeErrors]
-            if event_key in self.handlers and notify_expect_handler in self.handlers[event_key]:
-                self.handlers[event_key].remove(notify_expect_handler)
+            self.remove(event_type, notify_expect_handler)
 
     def _start(self) -> None:
         """Start the event bus if not already running"""

tests/test_remove_handler.py

@@ -0,0 +1,90 @@
+"""Tests for EventBus.remove() handler unsubscription."""
+
+from bubus import BaseEvent, EventBus
+
+
+class PingEvent(BaseEvent[str]):
+    """Minimal event used to verify handler removal."""
+
+
+async def test_remove_handler_by_event_class_stops_delivery():
+    """Removing a handler registered by event class stops it from receiving future events."""
+    bus = EventBus(name='remove_class')
+    calls: list[str] = []
+
+    async def handler(event: PingEvent) -> str:
+        calls.append(event.event_type)
+        return 'ok'
+
+    bus.on(PingEvent, handler)
+    assert bus.remove(PingEvent, handler) is True
+    assert bus.remove(PingEvent, handler) is False  # second removal is a no-op
+
+    await bus.dispatch(PingEvent())
+    await bus.wait_until_idle()
+
+    assert calls == []
+    await bus.stop()
+
+
+async def test_remove_handler_by_string_name():
+    """Removing a handler registered by event type string works too."""
+    bus = EventBus(name='remove_string')
+    calls: list[str] = []
+
+    async def handler(event: BaseEvent) -> str:
+        calls.append(event.event_type)
+        return 'ok'
+
+    bus.on('PingEvent', handler)
+    assert bus.remove('PingEvent', handler) is True
+    assert bus.remove('PingEvent', handler) is False
+
+    await bus.dispatch(PingEvent())
+    await bus.wait_until_idle()
+
+    assert calls == []
+    await bus.stop()
+
+
+async def test_remove_only_targets_specific_handler():
+    """Removing one handler leaves other handlers for the same event intact."""
+    bus = EventBus(name='remove_specific')
+    calls: list[str] = []
+
+    async def keep(event: PingEvent) -> str:
+        calls.append('keep')
+        return 'keep'
+
+    async def drop(event: PingEvent) -> str:
+        calls.append('drop')
+        return 'drop'
+
+    bus.on(PingEvent, keep)
+    bus.on(PingEvent, drop)
+    assert bus.remove(PingEvent, drop) is True
+
+    await bus.dispatch(PingEvent())
+    await bus.wait_until_idle()
+
+    assert calls == ['keep']
+    await bus.stop()
+
+
+async def test_remove_wildcard_handler():
+    """Removing a '*' wildcard handler stops it from receiving all events."""
+    bus = EventBus(name='remove_wildcard')
+    calls: list[str] = []
+
+    async def handler(event: BaseEvent) -> str:
+        calls.append(event.event_type)
+        return 'ok'
+
+    bus.on('*', handler)
+    assert bus.remove('*', handler) is True
+
+    await bus.dispatch(PingEvent())
+    await bus.wait_until_idle()
+
+    assert calls == []
+    await bus.stop()