Coverage for src / lexigram / contracts / exceptions / events.py: 0%

1"""Event bus and messaging exception classes.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6 

7from lexigram.contracts.exceptions.base import LexigramError 

8 

9 

10class EventError(LexigramError): 

11 """Base events/messaging error.""" 

12 

13 _code = "LEX_ERR_EVT_001" 

14 

15 def __init__(self, message: str = "Events error", **kwargs: Any) -> None: 

16 super().__init__(message, **kwargs) 

17 

18 

19class HandlerNotFoundError(EventError): 

20 """Event/command handler not found. 

21 

22 Raised when no handler is registered for a specific event or command type. 

23 """ 

24 

25 _code = "LEX_ERR_EVT_002" 

26 

27 def __init__( 

28 self, 

29 message: str = "Handler not found", 

30 handler_type: str | None = None, 

31 message_type: str | None = None, 

32 **kwargs: Any, 

33 ) -> None: 

34 details = kwargs.get("details", {}) 

35 if handler_type: 

36 details["handler_type"] = handler_type 

37 if message_type: 

38 details["message_type"] = message_type 

39 kwargs["details"] = details 

40 super().__init__(message, **kwargs) 

41 

42 

43__all__ = [ 

44 "DuplicateHandlerError", 

45 "EventError", 

46 "HandlerNotFoundError", 

47] 

48 

49 

50class DuplicateHandlerError(EventError): 

51 """Duplicate handler registration. 

52 

53 Raised when attempting to register a handler for a type that already has one. 

54 """ 

55 

56 _code = "LEX_ERR_EVT_003" 

57 

58 def __init__( 

59 self, 

60 message: str = "Duplicate handler", 

61 message_type: str | None = None, 

62 **kwargs: Any, 

63 ) -> None: 

64 details = kwargs.get("details", {}) 

65 if message_type: 

66 details["message_type"] = message_type 

67 kwargs["details"] = details 

68 super().__init__(message, **kwargs)