Coverage for polar/webhook/schemas.py: 100%

1from typing import Annotated 

2 

3from pydantic import UUID4, AnyUrl, Field, PlainSerializer, UrlConstraints 

4 

5from polar.kit.schemas import IDSchema, Schema, TimestampedSchema 

6from polar.models.webhook_endpoint import WebhookEventType, WebhookFormat 

7from polar.organization.schemas import OrganizationID 

8 

9HttpsUrl = Annotated[ 

10 AnyUrl, 

11 UrlConstraints( 

12 max_length=2083, 

13 allowed_schemes=["https"], 

14 host_required=True, 

15 ), 

16 PlainSerializer(lambda v: str(v), return_type=str), 

17] 

18 

19EndpointURL = Annotated[ 

20 HttpsUrl, 

21 Field( 

22 description="The URL where the webhook events will be sent.", 

23 examples=["https://webhook.site/cb791d80-f26e-4f8c-be88-6e56054192b0"], 

24 ), 

25] 

26EndpointFormat = Annotated[ 

27 WebhookFormat, 

28 Field(description="The format of the webhook payload."), 

29] 

30EndpointSecret = Annotated[ 

31 str, 

32 Field( 

33 description="The secret used to sign the webhook events.", 

34 examples=["polar_whs_ovyN6cPrTv56AApvzCaJno08SSmGJmgbWilb33N2JuK"], 

35 ), 

36] 

37EndpointEvents = Annotated[ 

38 list[WebhookEventType], 

39 Field(description="The events that will trigger the webhook."), 

40] 

41 

42 

43class WebhookEndpoint(IDSchema, TimestampedSchema): 

44 """ 

45 A webhook endpoint. 

46 """ 

47 

48 url: EndpointURL 

49 format: EndpointFormat 

50 secret: EndpointSecret 

51 organization_id: UUID4 = Field( 

52 description="The organization ID associated with the webhook endpoint." 

53 ) 

54 events: EndpointEvents 

55 enabled: bool = Field( 

56 description="Whether the webhook endpoint is enabled and will receive events." 

57 ) 

58 

59 

60class WebhookEndpointCreate(Schema): 

61 """ 

62 Schema to create a webhook endpoint. 

63 """ 

64 

65 url: EndpointURL 

66 secret: EndpointSecret | None = Field( 

67 default=None, 

68 deprecated="The secret is now generated on the backend.", 

69 min_length=32, 

70 ) 

71 format: EndpointFormat 

72 events: EndpointEvents 

73 organization_id: OrganizationID | None = Field( 

74 None, 

75 description=( 

76 "The organization ID associated with the webhook endpoint. " 

77 "**Required unless you use an organization token.**" 

78 ), 

79 ) 

80 

81 

82class WebhookEndpointUpdate(Schema): 

83 """ 

84 Schema to update a webhook endpoint. 

85 """ 

86 

87 url: EndpointURL | None = None 

88 secret: EndpointSecret | None = Field( 

89 default=None, 

90 deprecated="The secret should is now generated on the backend.", 

91 min_length=32, 

92 ) 

93 format: EndpointFormat | None = None 

94 events: EndpointEvents | None = None 

95 enabled: bool | None = Field( 

96 default=None, description="Whether the webhook endpoint is enabled." 

97 ) 

98 

99 

100class WebhookEvent(IDSchema, TimestampedSchema): 

101 """ 

102 A webhook event. 

103 

104 An event represent something that happened in the system 

105 that should be sent to the webhook endpoint. 

106 

107 It can be delivered multiple times until it's marked as succeeded, 

108 each one creating a new delivery. 

109 """ 

110 

111 last_http_code: int | None = Field( 

112 None, 

113 description="Last HTTP code returned by the URL. " 

114 "`null` if no delviery has been attempted or if the endpoint was unreachable.", 

115 ) 

116 succeeded: bool | None = Field( 

117 None, 

118 description=( 

119 "Whether this event was successfully delivered." 

120 " `null` if no delivery has been attempted." 

121 ), 

122 ) 

123 skipped: bool = Field( 

124 description="Whether this event was skipped because the webhook endpoint was disabled." 

125 ) 

126 payload: str | None = Field(description="The payload of the webhook event.") 

127 type: WebhookEventType = Field(description="The type of the webhook event.") 

128 is_archived: bool = Field( 

129 description=( 

130 "Whether this event is archived. " 

131 "Archived events can't be redelivered, " 

132 "and the payload is not accessible anymore." 

133 ), 

134 ) 

135 

136 

137class WebhookDelivery(IDSchema, TimestampedSchema): 

138 """ 

139 A webhook delivery for a webhook event. 

140 """ 

141 

142 succeeded: bool = Field(description="Whether the delivery was successful.") 

143 http_code: int | None = Field( 

144 description="The HTTP code returned by the URL." 

145 " `null` if the endpoint was unreachable.", 

146 ) 

147 response: str | None = Field( 

148 description=( 

149 "The response body returned by the URL, " 

150 "or the error message if the endpoint was unreachable." 

151 ), 

152 ) 

153 webhook_event: WebhookEvent = Field( 

154 description="The webhook event sent by this delivery." 

155 )