Skip to content
Merged
Show file tree
Hide file tree
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
3 changes: 3 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -901,3 +901,6 @@ FodyWeavers.xsd
# End of https://www.toptal.com/developers/gitignore/api/macos,linux,windows,python,jupyternotebooks,jetbrains,pycharm,vim,emacs,visualstudiocode,visualstudio

scratch/

# Allow resend/logs module (overrides [Ll]ogs/ rule above)
!resend/logs/
45 changes: 45 additions & 0 deletions examples/async/logs_async.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
import asyncio
import os

import resend

if not os.environ["RESEND_API_KEY"]:
raise EnvironmentError("RESEND_API_KEY is missing")


async def main() -> None:
logs: resend.Logs.ListResponse = await resend.Logs.list_async()
for log in logs["data"]:
print(log["id"])
print(log["endpoint"])
print(log["method"])
print(log["response_status"])
print(log["created_at"])

print("\n--- Using pagination parameters ---")
if logs["data"]:
paginated_params: resend.Logs.ListParams = {
"limit": 10,
"after": logs["data"][0]["id"],
}
paginated_logs: resend.Logs.ListResponse = await resend.Logs.list_async(
params=paginated_params
)
print(f"Retrieved {len(paginated_logs['data'])} logs with pagination")
print(f"Has more logs: {paginated_logs['has_more']}")
else:
print("No logs available for pagination example")

print("\n--- Retrieve a single log ---")
if logs["data"]:
log_id = logs["data"][0]["id"]
single_log: resend.Logs.GetResponse = await resend.Logs.get_async(log_id)
print(f"Log ID: {single_log['id']}")
print(f"Endpoint: {single_log['endpoint']}")
print(f"Method: {single_log['method']}")
print(f"Status: {single_log['response_status']}")
print(f"Request body: {single_log['request_body']}")
print(f"Response body: {single_log['response_body']}")


asyncio.run(main())
39 changes: 39 additions & 0 deletions examples/logs.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,39 @@
import os

import resend

if not os.environ["RESEND_API_KEY"]:
raise EnvironmentError("RESEND_API_KEY is missing")

logs: resend.Logs.ListResponse = resend.Logs.list()
for log in logs["data"]:
print(log["id"])
print(log["endpoint"])
print(log["method"])
print(log["response_status"])
print(log["created_at"])

print("\n--- Using pagination parameters ---")
if logs["data"]:
paginated_params: resend.Logs.ListParams = {
"limit": 10,
"after": logs["data"][0]["id"],
}
paginated_logs: resend.Logs.ListResponse = resend.Logs.list(
params=paginated_params
)
print(f"Retrieved {len(paginated_logs['data'])} logs with pagination")
print(f"Has more logs: {paginated_logs['has_more']}")
else:
print("No logs available for pagination example")

print("\n--- Retrieve a single log ---")
if logs["data"]:
log_id = logs["data"][0]["id"]
single_log: resend.Logs.GetResponse = resend.Logs.get(log_id)
print(f"Log ID: {single_log['id']}")
print(f"Endpoint: {single_log['endpoint']}")
print(f"Method: {single_log['method']}")
print(f"Status: {single_log['response_status']}")
print(f"Request body: {single_log['request_body']}")
print(f"Response body: {single_log['response_body']}")
4 changes: 4 additions & 0 deletions resend/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@
from .contacts.segments._contact_segments import ContactSegments
from .domains._domain import Domain
from .domains._domains import Domains
from .logs._log import Log
from .logs._logs import Logs
from .emails._attachment import Attachment, RemoteAttachment
from .emails._attachments import Attachments as EmailAttachments
from .emails._batch import Batch, BatchValidationError
Expand Down Expand Up @@ -74,6 +76,7 @@
"Templates",
"Webhooks",
"Topics",
"Logs",
# Types
"Audience",
"Contact",
Expand All @@ -84,6 +87,7 @@
"TopicSubscriptionUpdate",
"Domain",
"ApiKey",
"Log",
"Email",
"Attachment",
"RemoteAttachment",
Expand Down
4 changes: 4 additions & 0 deletions resend/logs/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
from resend.logs._log import Log
from resend.logs._logs import Logs

__all__ = ["Log", "Logs"]
38 changes: 38 additions & 0 deletions resend/logs/_log.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,38 @@
from typing import Any

from typing_extensions import NotRequired, TypedDict


class Log(TypedDict):
id: str
"""
The log ID
"""
created_at: str
"""
The date and time the log was created
"""
endpoint: str
"""
The API endpoint that was called
"""
method: str
"""
The HTTP method used
"""
response_status: int
"""
The HTTP response status code
"""
user_agent: str
"""
The user agent of the client
"""
request_body: NotRequired[Any]
"""
The original request body (only present when retrieving a single log)
"""
response_body: NotRequired[Any]
"""
The API response body (only present when retrieving a single log)
"""
193 changes: 193 additions & 0 deletions resend/logs/_logs.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,193 @@
from typing import Any, Dict, List, Optional, cast

from typing_extensions import NotRequired, TypedDict

from resend import request
from resend._base_response import BaseResponse
from resend.logs._log import Log
from resend.pagination_helper import PaginationHelper

# Async imports (optional - only available with pip install resend[async])
try:
from resend.async_request import AsyncRequest
except ImportError:
pass


class Logs:

class GetResponse(BaseResponse):
"""
GetResponse type that wraps a single log object

Attributes:
object (str): The object type, always "log"
id (str): The log ID
created_at (str): The date and time the log was created
endpoint (str): The API endpoint that was called
method (str): The HTTP method used
response_status (int): The HTTP response status code
user_agent (str): The user agent of the client
request_body (Any): The original request body
response_body (Any): The API response body
"""

object: str
"""
The object type, always "log"
"""
id: str
"""
The log ID
"""
created_at: str
"""
The date and time the log was created
"""
endpoint: str
"""
The API endpoint that was called
"""
method: str
"""
The HTTP method used
"""
response_status: int
"""
The HTTP response status code
"""
user_agent: str
"""
The user agent of the client
"""
request_body: Any
"""
The original request body
"""
response_body: Any
"""
The API response body
"""

class ListResponse(BaseResponse):
"""
ListResponse type that wraps a list of log objects with pagination metadata

Attributes:
object (str): The object type, always "list"
data (List[Log]): A list of log objects
has_more (bool): Whether there are more results available
"""

object: str
"""
The object type, always "list"
"""
data: List[Log]
"""
A list of log objects
"""
has_more: bool
"""
Whether there are more results available for pagination
"""

class ListParams(TypedDict):
limit: NotRequired[int]
"""
Number of logs to retrieve. Maximum is 100, and minimum is 1.
"""
after: NotRequired[str]
"""
The ID after which we'll retrieve more logs (for pagination).
This ID will not be included in the returned list.
Cannot be used with the before parameter.
"""
before: NotRequired[str]
"""
The ID before which we'll retrieve more logs (for pagination).
This ID will not be included in the returned list.
Cannot be used with the after parameter.
"""

@classmethod
def get(cls, log_id: str) -> GetResponse:
"""
Retrieve a single log by its ID.
see more: https://resend.com/docs/api-reference/logs/retrieve-log

Args:
log_id (str): The ID of the log to retrieve

Returns:
GetResponse: The log object
"""
path = f"/logs/{log_id}"
resp = request.Request[Logs.GetResponse](
path=path, params={}, verb="get"
).perform_with_content()
return resp

@classmethod
def list(cls, params: Optional[ListParams] = None) -> ListResponse:
"""
Retrieve a list of logs.
see more: https://resend.com/docs/api-reference/logs/list-logs

Args:
params (Optional[ListParams]): Optional pagination parameters
- limit: Number of logs to retrieve (max 100, min 1)
- after: ID after which to retrieve more logs
- before: ID before which to retrieve more logs

Returns:
ListResponse: A list of log objects
"""
base_path = "/logs"
query_params = cast(Dict[Any, Any], params) if params else None
path = PaginationHelper.build_paginated_path(base_path, query_params)
resp = request.Request[Logs.ListResponse](
path=path, params={}, verb="get"
).perform_with_content()
return resp

@classmethod
async def get_async(cls, log_id: str) -> GetResponse:
"""
Retrieve a single log by its ID (async).
see more: https://resend.com/docs/api-reference/logs/retrieve-log

Args:
log_id (str): The ID of the log to retrieve

Returns:
GetResponse: The log object
"""
path = f"/logs/{log_id}"
resp = await AsyncRequest[Logs.GetResponse](
path=path, params={}, verb="get"
).perform_with_content()
return resp

@classmethod
async def list_async(cls, params: Optional[ListParams] = None) -> ListResponse:
"""
Retrieve a list of logs (async).
see more: https://resend.com/docs/api-reference/logs/list-logs

Args:
params (Optional[ListParams]): Optional pagination parameters
- limit: Number of logs to retrieve (max 100, min 1)
- after: ID after which to retrieve more logs
- before: ID before which to retrieve more logs

Returns:
ListResponse: A list of log objects
"""
base_path = "/logs"
query_params = cast(Dict[Any, Any], params) if params else None
path = PaginationHelper.build_paginated_path(base_path, query_params)
resp = await AsyncRequest[Logs.ListResponse](
path=path, params={}, verb="get"
).perform_with_content()
return resp
2 changes: 1 addition & 1 deletion resend/version.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
__version__ = "2.26.0"
__version__ = "2.27.0"


def get_version() -> str:
Expand Down
Loading
Loading