Coverage for polar/webhook/schemas.py: 100%
41 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-12-05 16:17 +0000
1from typing import Annotated 1a
3from pydantic import UUID4, AnyUrl, Field, PlainSerializer, UrlConstraints 1a
5from polar.kit.schemas import IDSchema, Schema, TimestampedSchema 1a
6from polar.models.webhook_endpoint import WebhookEventType, WebhookFormat 1a
7from polar.organization.schemas import OrganizationID 1a
9HttpsUrl = Annotated[ 1a
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]
19EndpointURL = Annotated[ 1a
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[ 1a
27 WebhookFormat,
28 Field(description="The format of the webhook payload."),
29]
30EndpointSecret = Annotated[ 1a
31 str,
32 Field(
33 description="The secret used to sign the webhook events.",
34 examples=["polar_whs_ovyN6cPrTv56AApvzCaJno08SSmGJmgbWilb33N2JuK"],
35 ),
36]
37EndpointEvents = Annotated[ 1a
38 list[WebhookEventType],
39 Field(description="The events that will trigger the webhook."),
40]
43class WebhookEndpoint(IDSchema, TimestampedSchema): 1a
44 """
45 A webhook endpoint.
46 """
48 url: EndpointURL 1a
49 format: EndpointFormat 1a
50 secret: EndpointSecret 1a
51 organization_id: UUID4 = Field( 1a
52 description="The organization ID associated with the webhook endpoint."
53 )
54 events: EndpointEvents 1a
55 enabled: bool = Field( 1a
56 description="Whether the webhook endpoint is enabled and will receive events."
57 )
60class WebhookEndpointCreate(Schema): 1a
61 """
62 Schema to create a webhook endpoint.
63 """
65 url: EndpointURL 1a
66 secret: EndpointSecret | None = Field( 1a
67 default=None,
68 deprecated="The secret is now generated on the backend.",
69 min_length=32,
70 )
71 format: EndpointFormat 1a
72 events: EndpointEvents 1a
73 organization_id: OrganizationID | None = Field( 1a
74 None,
75 description=(
76 "The organization ID associated with the webhook endpoint. "
77 "**Required unless you use an organization token.**"
78 ),
79 )
82class WebhookEndpointUpdate(Schema): 1a
83 """
84 Schema to update a webhook endpoint.
85 """
87 url: EndpointURL | None = None 1a
88 secret: EndpointSecret | None = Field( 1a
89 default=None,
90 deprecated="The secret should is now generated on the backend.",
91 min_length=32,
92 )
93 format: EndpointFormat | None = None 1a
94 events: EndpointEvents | None = None 1a
95 enabled: bool | None = Field( 1a
96 default=None, description="Whether the webhook endpoint is enabled."
97 )
100class WebhookEvent(IDSchema, TimestampedSchema): 1a
101 """
102 A webhook event.
104 An event represent something that happened in the system
105 that should be sent to the webhook endpoint.
107 It can be delivered multiple times until it's marked as succeeded,
108 each one creating a new delivery.
109 """
111 last_http_code: int | None = Field( 1a
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( 1a
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( 1a
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.") 1a
127 type: WebhookEventType = Field(description="The type of the webhook event.") 1a
128 is_archived: bool = Field( 1a
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 )
137class WebhookDelivery(IDSchema, TimestampedSchema): 1a
138 """
139 A webhook delivery for a webhook event.
140 """
142 succeeded: bool = Field(description="Whether the delivery was successful.") 1a
143 http_code: int | None = Field( 1a
144 description="The HTTP code returned by the URL."
145 " `null` if the endpoint was unreachable.",
146 )
147 response: str | None = Field( 1a
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( 1a
154 description="The webhook event sent by this delivery."
155 )