Coverage for src/lexigram/web/routing/cqrs.py: 53%

1"""CQRSController - Controller base class with integrated CQRS command/query bus dispatch. 

2 

3Provides a controller that can dispatch commands and queries through the CQRS buses 

4registered in the DI container. Controllers extending this class get typed 

5``dispatch_command`` and ``dispatch_query`` helper methods. 

6""" 

7 

8from __future__ import annotations 

9 

10from typing import Any 

11 

12from lexigram.contracts.events import CommandBusProtocol, QueryBusProtocol 

13from lexigram.logging import get_logger 

14from lexigram.web.routing.controllers import Controller 

15 

16logger = get_logger(__name__) 

17 

18 

19class CQRSController(Controller): 

20 """Controller with built-in CQRS command and query bus dispatch. 

21 

22 Subclass this instead of ``Controller`` when your endpoints need to 

23 dispatch commands or queries through the CQRS buses. 

24 

25 The buses are injected via the constructor and are expected to be 

26 resolved from the DI container. 

27 

28 Example:: 

29 

30 class OrderController(CQRSController): 

31 @post("/orders") 

32 async def create_order(self, request: Request) -> JSONResponse: 

33 data = await request.json() 

34 result = await self.dispatch_command(CreateOrderCommand(**data)) 

35 return JSONResponse({"id": result.id}, status_code=201) 

36 

37 @get("/orders/{order_id}") 

38 async def get_order(self, request: Request) -> JSONResponse: 

39 order_id = request.path_params["order_id"] 

40 result = await self.dispatch_query(GetOrderQuery(order_id=order_id)) 

41 return JSONResponse(result.to_dict()) 

42 """ 

43 

44 def __init__( 

45 self, 

46 command_bus: CommandBusProtocol | None = None, 

47 query_bus: QueryBusProtocol | None = None, 

48 ) -> None: 

49 """Initialize CQRSController with optional command and query buses. 

50 

51 Args: 

52 command_bus: CommandBusProtocol implementation for dispatching commands. 

53 If None, calling ``dispatch_command`` will raise RuntimeError. 

54 query_bus: QueryBusProtocol implementation for executing queries. 

55 If None, calling ``dispatch_query`` will raise RuntimeError. 

56 """ 

57 super().__init__() 

58 self._command_bus = command_bus 

59 self._query_bus = query_bus 

60 

61 async def dispatch_command(self, command: Any) -> Any: 

62 """Dispatch a command through the command bus. 

63 

64 Args: 

65 command: The command object to dispatch. 

66 

67 Returns: 

68 The result returned by the command handler. 

69 

70 Raises: 

71 RuntimeError: If no CommandBusProtocol was provided to this controller. 

72 """ 

73 if self._command_bus is None: 

74 raise RuntimeError( 

75 f"{type(self).__name__} has no CommandBusProtocol configured. " 

76 "Register a CommandBusProtocol in the DI container.", 

77 ) 

78 return await self._command_bus.dispatch(command) 

79 

80 async def dispatch_query(self, query: Any) -> Any: 

81 """Dispatch a query through the query bus. 

82 

83 Args: 

84 query: The query object to execute. 

85 

86 Returns: 

87 The query result. 

88 

89 Raises: 

90 RuntimeError: If no QueryBusProtocol was provided to this controller. 

91 """ 

92 if self._query_bus is None: 

93 raise RuntimeError( 

94 f"{type(self).__name__} has no QueryBusProtocol configured. " 

95 "Register a QueryBusProtocol in the DI container.", 

96 ) 

97 return await self._query_bus.execute(query)