Skip to content
Draft
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 68 additions & 0 deletions resend/webhooks/_webhooks.py
Original file line number Diff line number Diff line change
Expand Up @@ -363,6 +363,17 @@ def list(cls, params: Optional[ListParams] = None) -> ListResponse:
def list_events(
cls, webhook_id: str, params: Optional[ListEventsParams] = None
) -> ListEventsResponse:
"""
Retrieve a list of events delivered to a webhook.
see more: https://resend.com/docs/api-reference/webhooks/list-events

Args:
webhook_id (str): The webhook ID
params (Optional[ListEventsParams]): Optional pagination parameters

Returns:
ListEventsResponse: A list of webhook events
"""
base_path = f"/webhooks/{webhook_id}/events"
query_params = cast(Dict[Any, Any], params) if params else None
path = PaginationHelper.build_paginated_path(base_path, query_params)
Expand All @@ -372,6 +383,17 @@ def list_events(

@classmethod
def get_event(cls, webhook_id: str, event_id: str) -> GetEventResponse:
"""
Retrieve a single event delivered to a webhook.
see more: https://resend.com/docs/api-reference/webhooks/get-event

Args:
webhook_id (str): The webhook ID
event_id (str): The webhook event ID

Returns:
GetEventResponse: The webhook event
"""
path = f"/webhooks/{webhook_id}/events/{event_id}"
return request.Request[Webhooks.GetEventResponse](
path=path, params={}, verb="get"
Expand All @@ -384,6 +406,18 @@ def list_event_attempts(
event_id: str,
params: Optional[ListEventAttemptsParams] = None,
) -> ListEventAttemptsResponse:
"""
Retrieve the delivery attempts for a webhook event.
see more: https://resend.com/docs/api-reference/webhooks/list-event-attempts

Args:
webhook_id (str): The webhook ID
event_id (str): The webhook event ID
params (Optional[ListEventAttemptsParams]): Optional pagination parameters

Returns:
ListEventAttemptsResponse: A list of delivery attempts
"""
base_path = f"/webhooks/{webhook_id}/events/{event_id}/attempts"
query_params = cast(Dict[Any, Any], params) if params else None
path = PaginationHelper.build_paginated_path(base_path, query_params)
Expand Down Expand Up @@ -588,6 +622,17 @@ async def list_async(cls, params: Optional[ListParams] = None) -> ListResponse:
async def list_events_async(
cls, webhook_id: str, params: Optional[ListEventsParams] = None
) -> ListEventsResponse:
"""
Retrieve a list of events delivered to a webhook (async).
see more: https://resend.com/docs/api-reference/webhooks/list-events

Args:
webhook_id (str): The webhook ID
params (Optional[ListEventsParams]): Optional pagination parameters

Returns:
ListEventsResponse: A list of webhook events
"""
base_path = f"/webhooks/{webhook_id}/events"
query_params = cast(Dict[Any, Any], params) if params else None
path = PaginationHelper.build_paginated_path(base_path, query_params)
Expand All @@ -597,6 +642,17 @@ async def list_events_async(

@classmethod
async def get_event_async(cls, webhook_id: str, event_id: str) -> GetEventResponse:
"""
Retrieve a single event delivered to a webhook (async).
see more: https://resend.com/docs/api-reference/webhooks/get-event

Args:
webhook_id (str): The webhook ID
event_id (str): The webhook event ID

Returns:
GetEventResponse: The webhook event
"""
path = f"/webhooks/{webhook_id}/events/{event_id}"
return await AsyncRequest[Webhooks.GetEventResponse](
path=path, params={}, verb="get"
Expand All @@ -609,6 +665,18 @@ async def list_event_attempts_async(
event_id: str,
params: Optional[ListEventAttemptsParams] = None,
) -> ListEventAttemptsResponse:
"""
Retrieve the delivery attempts for a webhook event (async).
see more: https://resend.com/docs/api-reference/webhooks/list-event-attempts

Args:
webhook_id (str): The webhook ID
event_id (str): The webhook event ID
params (Optional[ListEventAttemptsParams]): Optional pagination parameters

Returns:
ListEventAttemptsResponse: A list of delivery attempts
"""
base_path = f"/webhooks/{webhook_id}/events/{event_id}/attempts"
query_params = cast(Dict[Any, Any], params) if params else None
path = PaginationHelper.build_paginated_path(base_path, query_params)
Expand Down
Loading