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

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

2"""Queue value types.""" 

3 

4from __future__ import annotations 

5 

6from dataclasses import dataclass, field 

7from enum import StrEnum 

8import time 

9from typing import Any 

10import uuid 

11 

12 

13class DeliveryGuarantee(StrEnum): 

14 """Message delivery guarantees supported by queue backends. 

15 

16 Attributes: 

17 AT_MOST_ONCE: Message may be lost but is never duplicated. 

18 AT_LEAST_ONCE: Message may be duplicated but is never lost. 

19 EXACTLY_ONCE: Message is delivered exactly once (best effort). 

20 """ 

21 

22 AT_MOST_ONCE = "at_most_once" 

23 AT_LEAST_ONCE = "at_least_once" 

24 EXACTLY_ONCE = "exactly_once" 

25 

26 

27@dataclass(frozen=True) 

28class BusMessage: 

29 """A message transported through the queue backend. 

30 

31 ``id`` defaults to a new UUID4. ``timestamp`` defaults to the current 

32 wall-clock time. Both are set via field defaults so callers do not need 

33 to supply them. 

34 

35 Attributes: 

36 id: Unique message identifier. 

37 topic: Destination topic or queue name. 

38 payload: Arbitrary message body; must be serialisable by the backend. 

39 headers: Opaque string key-value headers passed to the broker. 

40 timestamp: Unix epoch timestamp when the message was created. 

41 ttl: Time-to-live in seconds; ``None`` uses the backend default. 

42 priority: Message priority (higher = more urgent); backend-dependent. 

43 delivery_guarantee: Delivery semantics for this message. 

44 retry_count: Number of times this message has been retried. 

45 max_retries: Maximum retries before routing to DLQ. 

46 """ 

47 

48 id: str = field(default_factory=lambda: str(uuid.uuid4())) 

49 topic: str = "" 

50 payload: Any = None 

51 headers: dict[str, str] = field(default_factory=dict) 

52 timestamp: float = field(default_factory=time.time) 

53 ttl: float | None = None 

54 priority: int = 0 

55 delivery_guarantee: DeliveryGuarantee = DeliveryGuarantee.AT_LEAST_ONCE 

56 retry_count: int = 0 

57 max_retries: int = 3 

58 

59 def is_expired(self) -> bool: 

60 """Return ``True`` if the message TTL has elapsed.""" 

61 if self.ttl is None: 

62 return False 

63 return time.time() > self.timestamp + self.ttl 

64 

65 def should_retry(self) -> bool: 

66 """Return ``True`` if the message may be retried.""" 

67 return self.retry_count < self.max_retries and not self.is_expired() 

68 

69 

70__all__ = ["BusMessage", "DeliveryGuarantee"]