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

1from typing import Annotated 

2 

3from fastapi import Depends, Path, Query 

4from pydantic import UUID4, AwareDatetime 

5 

6from polar.exceptions import ResourceNotFound 

7from polar.kit.pagination import ListResource, PaginationParamsQuery 

8from polar.kit.schemas import MultipleQueryFilter 

9from polar.models import WebhookEndpoint 

10from polar.openapi import APITag 

11from polar.organization.schemas import OrganizationID 

12from polar.postgres import AsyncSession, get_db_session 

13from polar.routing import APIRouter 

14 

15from .auth import WebhooksRead, WebhooksWrite 

16from .schemas import WebhookDelivery as WebhookDeliverySchema 

17from .schemas import WebhookEndpoint as WebhookEndpointSchema 

18from .schemas import WebhookEndpointCreate, WebhookEndpointUpdate 

19from .service import webhook as webhook_service 

20 

21router = APIRouter(prefix="/webhooks", tags=["webhooks", APITag.public]) 

22 

23WebhookEndpointID = Annotated[UUID4, Path(description="The webhook endpoint ID.")] 

24WebhookEndpointNotFound = { 

25 "description": "Webhook endpoint not found.", 

26 "model": ResourceNotFound.schema(), 

27} 

28 

29 

30@router.get("/endpoints", response_model=ListResource[WebhookEndpointSchema]) 

31async def list_webhook_endpoints( 

32 pagination: PaginationParamsQuery, 

33 auth_subject: WebhooksRead, 

34 organization_id: MultipleQueryFilter[OrganizationID] | None = Query( 

35 None, description="Filter by organization ID." 

36 ), 

37 session: AsyncSession = Depends(get_db_session), 

38) -> ListResource[WebhookEndpointSchema]: 

39 """List webhook endpoints.""" 

40 results, count = await webhook_service.list_endpoints( 

41 session, 

42 auth_subject, 

43 organization_id=organization_id, 

44 pagination=pagination, 

45 ) 

46 return ListResource.from_paginated_results( 

47 [WebhookEndpointSchema.model_validate(result) for result in results], 

48 count, 

49 pagination, 

50 ) 

51 

52 

53@router.get( 

54 "/endpoints/{id}", 

55 response_model=WebhookEndpointSchema, 

56 responses={404: WebhookEndpointNotFound}, 

57) 

58async def get_webhook_endpoint( 

59 id: WebhookEndpointID, 

60 auth_subject: WebhooksRead, 

61 session: AsyncSession = Depends(get_db_session), 

62) -> WebhookEndpoint: 

63 """Get a webhook endpoint by ID.""" 

64 endpoint = await webhook_service.get_endpoint(session, auth_subject, id) 

65 if not endpoint: 

66 raise ResourceNotFound() 

67 

68 return endpoint 

69 

70 

71@router.post( 

72 "/endpoints", 

73 response_model=WebhookEndpointSchema, 

74 status_code=201, 

75 responses={201: {"description": "Webhook endpoint created."}}, 

76) 

77async def create_webhook_endpoint( 

78 endpoint_create: WebhookEndpointCreate, 

79 auth_subject: WebhooksWrite, 

80 session: AsyncSession = Depends(get_db_session), 

81) -> WebhookEndpoint: 

82 """ 

83 Create a webhook endpoint. 

84 """ 

85 return await webhook_service.create_endpoint(session, auth_subject, endpoint_create) 

86 

87 

88@router.patch( 

89 "/endpoints/{id}", 

90 response_model=WebhookEndpointSchema, 

91 responses={ 

92 200: {"description": "Webhook endpoint updated."}, 

93 404: WebhookEndpointNotFound, 

94 }, 

95) 

96async def update_webhook_endpoint( 

97 id: WebhookEndpointID, 

98 update: WebhookEndpointUpdate, 

99 auth_subject: WebhooksWrite, 

100 session: AsyncSession = Depends(get_db_session), 

101) -> WebhookEndpoint: 

102 """ 

103 Update a webhook endpoint. 

104 """ 

105 endpoint = await webhook_service.get_endpoint(session, auth_subject, id) 

106 if not endpoint: 

107 raise ResourceNotFound() 

108 

109 return await webhook_service.update_endpoint( 

110 session, endpoint=endpoint, update_schema=update 

111 ) 

112 

113 

114@router.patch( 

115 "/endpoints/{id}/secret", 

116 response_model=WebhookEndpointSchema, 

117 responses={ 

118 200: {"description": "Webhook endpoint secret reset."}, 

119 404: WebhookEndpointNotFound, 

120 }, 

121) 

122async def reset_webhook_endpoint_secret( 

123 id: WebhookEndpointID, 

124 auth_subject: WebhooksWrite, 

125 session: AsyncSession = Depends(get_db_session), 

126) -> WebhookEndpoint: 

127 """ 

128 Regenerate a webhook endpoint secret. 

129 """ 

130 endpoint = await webhook_service.get_endpoint(session, auth_subject, id) 

131 if not endpoint: 

132 raise ResourceNotFound() 

133 

134 return await webhook_service.reset_endpoint_secret(session, endpoint=endpoint) 

135 

136 

137@router.delete( 

138 "/endpoints/{id}", 

139 status_code=204, 

140 responses={ 

141 204: {"description": "Webhook endpoint deleted."}, 

142 404: WebhookEndpointNotFound, 

143 }, 

144) 

145async def delete_webhook_endpoint( 

146 id: WebhookEndpointID, 

147 auth_subject: WebhooksWrite, 

148 session: AsyncSession = Depends(get_db_session), 

149) -> None: 

150 """ 

151 Delete a webhook endpoint. 

152 """ 

153 endpoint = await webhook_service.get_endpoint(session, auth_subject, id) 

154 if not endpoint: 

155 raise ResourceNotFound() 

156 

157 await webhook_service.delete_endpoint(session, endpoint) 

158 

159 

160@router.get( 

161 "/deliveries", 

162 response_model=ListResource[WebhookDeliverySchema], 

163) 

164async def list_webhook_deliveries( 

165 pagination: PaginationParamsQuery, 

166 auth_subject: WebhooksRead, 

167 endpoint_id: MultipleQueryFilter[UUID4] | None = Query( 

168 None, description="Filter by webhook endpoint ID." 

169 ), 

170 start_timestamp: AwareDatetime | None = Query( 

171 None, description="Filter deliveries after this timestamp." 

172 ), 

173 end_timestamp: AwareDatetime | None = Query( 

174 None, description="Filter deliveries before this timestamp." 

175 ), 

176 session: AsyncSession = Depends(get_db_session), 

177) -> ListResource[WebhookDeliverySchema]: 

178 """ 

179 List webhook deliveries. 

180 

181 Deliveries are all the attempts to deliver a webhook event to an endpoint. 

182 """ 

183 results, count = await webhook_service.list_deliveries( 

184 session, 

185 auth_subject, 

186 endpoint_id=endpoint_id, 

187 start_timestamp=start_timestamp, 

188 end_timestamp=end_timestamp, 

189 pagination=pagination, 

190 ) 

191 

192 return ListResource.from_paginated_results( 

193 [WebhookDeliverySchema.model_validate(result) for result in results], 

194 count, 

195 pagination, 

196 ) 

197 

198 

199@router.post( 

200 "/events/{id}/redeliver", 

201 status_code=202, 

202 responses={ 

203 202: {"description": "Webhook event re-delivery scheduled."}, 

204 404: { 

205 "description": "Webhook event not found.", 

206 "model": ResourceNotFound.schema(), 

207 }, 

208 }, 

209) 

210async def redeliver_webhook_event( 

211 id: Annotated[UUID4, Path(..., description="The webhook event ID.")], 

212 auth_subject: WebhooksWrite, 

213 session: AsyncSession = Depends(get_db_session), 

214) -> None: 

215 """ 

216 Schedule the re-delivery of a webhook event. 

217 """ 

218 return await webhook_service.redeliver_event(session, auth_subject, id)