Coverage for src / lexigram / contracts / webhook / types.py: 0%

1"""Webhook domain types: subscriptions, events, and delivery attempts.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass, field 

6from datetime import UTC, datetime 

7from enum import Enum 

8from typing import Any 

9 

10 

11class DeliveryStatus(str, Enum): 

12 """Status of a webhook delivery attempt.""" 

13 

14 PENDING = "pending" 

15 DELIVERED = "delivered" 

16 FAILED = "failed" 

17 DEAD_LETTER = "dead_letter" 

18 CANCELLED = "cancelled" 

19 

20 

21@dataclass(frozen=True) 

22class WebhookSubscription: 

23 """A registered webhook endpoint subscription. 

24 

25 Attributes: 

26 subscription_id: Unique identifier (UUID). 

27 url: HTTP(S) endpoint URL to deliver events to. 

28 secret: Shared secret for HMAC signature computation. 

29 event_types: Set of event type names to deliver. None = all events. 

30 active: Whether the subscription is currently active. 

31 description: Human-readable label for the subscription. 

32 tenant_id: Optional multi-tenant scoping. 

33 created_at: When the subscription was created. 

34 metadata: Arbitrary key-value context. 

35 """ 

36 

37 subscription_id: str 

38 url: str 

39 secret: str 

40 event_types: frozenset[str] | None = None 

41 active: bool = True 

42 description: str = "" 

43 tenant_id: str | None = None 

44 created_at: datetime = field(default_factory=lambda: datetime.now(UTC)) 

45 metadata: dict[str, Any] = field(default_factory=dict) 

46 

47 def __repr__(self) -> str: 

48 return ( 

49 f"WebhookSubscription(subscription_id={self.subscription_id!r}, " 

50 f"url={self.url!r}, secret='***', event_types={self.event_types!r}, " 

51 f"active={self.active!r}, description={self.description!r}, " 

52 f"tenant_id={self.tenant_id!r}, created_at={self.created_at!r}, " 

53 f"metadata={self.metadata!r})" 

54 ) 

55 

56 

57@dataclass(frozen=True) 

58class WebhookEvent: 

59 """An event prepared for webhook delivery. 

60 

61 Attributes: 

62 event_id: Unique identifier for deduplication (UUID). 

63 event_type: Dot-notation event type (e.g. "user.created"). 

64 payload: JSON-serializable event data. 

65 occurred_at: When the original event occurred. 

66 source: Originating service or module. 

67 """ 

68 

69 event_id: str 

70 event_type: str 

71 payload: dict[str, Any] 

72 occurred_at: datetime = field(default_factory=lambda: datetime.now(UTC)) 

73 source: str = "" 

74 

75 

76@dataclass(frozen=True) 

77class DeliveryAttempt: 

78 """Record of a single webhook delivery attempt. 

79 

80 Attributes: 

81 attempt_id: Unique identifier for this attempt. 

82 subscription_id: Target subscription. 

83 event_id: The event being delivered. 

84 event_type: Event type name. 

85 status: Outcome of this attempt. 

86 status_code: HTTP response status code (None if connection failed). 

87 attempt_number: 1-based attempt counter. 

88 attempted_at: When this attempt was made. 

89 next_retry_at: Scheduled time for the next retry (None if terminal). 

90 error_message: Error details on failure. 

91 duration_ms: Round-trip time in milliseconds. 

92 """ 

93 

94 attempt_id: str 

95 subscription_id: str 

96 event_id: str 

97 event_type: str 

98 status: DeliveryStatus 

99 status_code: int | None = None 

100 attempt_number: int = 1 

101 attempted_at: datetime = field(default_factory=lambda: datetime.now(UTC)) 

102 next_retry_at: datetime | None = None 

103 error_message: str | None = None 

104 duration_ms: float | None = None 

105 

106 

107__all__ = [ 

108 "DeliveryAttempt", 

109 "DeliveryStatus", 

110 "WebhookEvent", 

111 "WebhookSubscription", 

112]