Coverage for src / lexigram / contracts / queue / protocols.py: 100%

1# lexigram-contracts/src/lexigram/contracts/queue/protocols.py 

2"""Queue and consumer protocols.""" 

3 

4from __future__ import annotations 

5 

6from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable 

7 

8if TYPE_CHECKING: 

9 from collections.abc import Callable, Coroutine 

10 

11 from lexigram.contracts.core.health import HealthCheckResult 

12 from lexigram.contracts.queue.types import BusMessage 

13 

14 

15@runtime_checkable 

16class QueueProtocol(Protocol): 

17 """Structural protocol for message queue / bus backends. 

18 

19 Implementations include in-memory, Redis Pub/Sub, RabbitMQ, Kafka, 

20 and SQS. ``connect()`` / ``close()`` manage the connection lifecycle; 

21 ``publish()`` and ``subscribe()`` handle message delivery. 

22 """ 

23 

24 async def connect(self) -> None: 

25 """Establish connection to the broker.""" 

26 ... 

27 

28 async def close(self) -> None: 

29 """Close the connection and release resources.""" 

30 ... 

31 

32 async def publish(self, topic: str, message: BusMessage) -> None: 

33 """Publish a message to a topic. 

34 

35 Args: 

36 topic: Destination topic or queue name. 

37 message: The message to publish. 

38 

39 Raises: 

40 QueueError: For expected publish failures (quota, auth, etc.). 

41 """ 

42 ... 

43 

44 async def subscribe( 

45 self, 

46 topic: str, 

47 handler: Callable[[BusMessage], Coroutine[Any, Any, None]], 

48 ) -> None: 

49 """Subscribe a handler to a topic. 

50 

51 The handler is called for each message received. The backend is 

52 responsible for acknowledging or rejecting messages. 

53 

54 Args: 

55 topic: Topic or queue name to subscribe to. 

56 handler: Async callable invoked per message. 

57 """ 

58 ... 

59 

60 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

61 """Check broker connectivity.""" 

62 ... 

63 

64 

65@runtime_checkable 

66class MessageConsumerProtocol(Protocol): 

67 """Protocol for consumer worker classes. 

68 

69 Consumer workers register topic subscriptions and run the event loop. 

70 """ 

71 

72 async def start(self) -> None: 

73 """Start consuming messages from subscribed topics.""" 

74 ... 

75 

76 async def stop(self) -> None: 

77 """Stop consuming and drain in-flight messages.""" 

78 ... 

79 

80 

81__all__ = ["MessageConsumerProtocol", "QueueProtocol"]