Repository navigation
Expand file tree
/
Copy pathbytesend.py
More file actions
195 lines (162 loc) · 6.43 KB
/
Copy pathbytesend.py
File metadata and controls
195 lines (162 loc) · 6.43 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
"""Core client for interacting with the ByteSend API.
Enhancements:
- Optional ``raise_on_error`` to raise ``ByteSendHTTPError`` on non-2xx.
- Reusable ``requests.Session`` support for connection reuse.
"""
from __future__ import annotations
import os
from typing import Any, Dict, Optional, Tuple
import requests
DEFAULT_BASE_URL = "https://bytesend.cloud"
class ByteSendHTTPError(Exception):
"""HTTP error raised when ``raise_on_error=True`` and a request fails."""
def __init__(self, status_code: int, error: Dict[str, Any], method: str, path: str) -> None:
self.status_code = status_code
self.error = error
self.method = method
self.path = path
super().__init__(self.__str__())
def __str__(self) -> str: # pragma: no cover - presentation only
code = self.error.get("code", "UNKNOWN_ERROR")
message = self.error.get("message", "")
return f"{self.method} {self.path} -> {self.status_code} {code}: {message}"
class ByteSend:
"""ByteSend API client.
Parameters
----------
key:
API key issued by ByteSend. If not provided, the client attempts to
read ``BYTESEND_API_KEY``
from the environment.
url:
Optional base URL for the API (useful for testing).
"""
def __init__(
self,
key: Optional[str] = None,
url: Optional[str] = None,
*,
raise_on_error: bool = True,
session: Optional[requests.Session] = None,
) -> None:
self.key = key or os.getenv("BYTESEND_API_KEY")
if not self.key:
raise ValueError("Missing API key. Pass it to ByteSend('bs_123')")
base = os.getenv("BYTESEND_BASE_URL") or DEFAULT_BASE_URL
if url:
base = url
self.url = f"{base}/api/v1"
self.headers = {
"Authorization": f"Bearer {self.key}",
"Content-Type": "application/json",
}
self.raise_on_error = raise_on_error
self._session = session or requests.Session()
# Lazily initialise resource clients.
self.emails = Emails(self)
self.contacts = Contacts(self)
self.contact_books = ContactBooks(self)
self.contactBooks = self.contact_books
self.domains = Domains(self)
self.campaigns = Campaigns(self)
# ------------------------------------------------------------------
# Webhooks
# ------------------------------------------------------------------
def webhooks(self, secret: str) -> "Webhooks":
"""Create a Webhooks instance for verifying webhook signatures.
Parameters
----------
secret:
The webhook signing secret (starts with 'whsec_').
Returns
-------
Webhooks
A Webhooks instance for verifying signatures and constructing events.
Example
-------
```python
bytesend = ByteSend("us_12345")
webhooks = bytesend.webhooks("whsec_xxx")
# In your webhook handler
event = webhooks.construct_event(body, headers=request.headers)
```
"""
return Webhooks(secret)
# ------------------------------------------------------------------
# Internal request helper
# ------------------------------------------------------------------
def _build_headers(self, extra: Optional[Dict[str, str]] = None) -> Dict[str, str]:
headers = dict(self.headers)
if extra:
headers.update({k: v for k, v in extra.items() if v is not None})
return headers
def _request(
self,
method: str,
path: str,
json: Optional[Any] = None,
headers: Optional[Dict[str, str]] = None,
) -> Tuple[Optional[Dict[str, Any]], Optional[Dict[str, Any]]]:
"""Perform an HTTP request and return ``(data, error)``."""
resp = self._session.request(
method,
f"{self.url}{path}",
headers=self._build_headers(headers),
json=json,
)
default_error = {"code": "INTERNAL_SERVER_ERROR", "message": resp.reason}
if not resp.ok:
try:
payload = resp.json()
error = payload.get("error", default_error)
except Exception:
error = default_error
if self.raise_on_error:
raise ByteSendHTTPError(resp.status_code, error, method, path)
return None, error
try:
return resp.json(), None
except Exception:
return None, default_error
# ------------------------------------------------------------------
# HTTP verb helpers
# ------------------------------------------------------------------
def post(
self,
path: str,
body: Any,
headers: Optional[Dict[str, str]] = None,
) -> Tuple[Optional[Dict[str, Any]], Optional[Dict[str, Any]]]:
return self._request("POST", path, json=body, headers=headers)
def get(
self, path: str, headers: Optional[Dict[str, str]] = None
) -> Tuple[Optional[Dict[str, Any]], Optional[Dict[str, Any]]]:
return self._request("GET", path, headers=headers)
def put(
self,
path: str,
body: Any,
headers: Optional[Dict[str, str]] = None,
) -> Tuple[Optional[Dict[str, Any]], Optional[Dict[str, Any]]]:
return self._request("PUT", path, json=body, headers=headers)
def patch(
self,
path: str,
body: Any,
headers: Optional[Dict[str, str]] = None,
) -> Tuple[Optional[Dict[str, Any]], Optional[Dict[str, Any]]]:
return self._request("PATCH", path, json=body, headers=headers)
def delete(
self,
path: str,
body: Optional[Any] = None,
headers: Optional[Dict[str, str]] = None,
) -> Tuple[Optional[Dict[str, Any]], Optional[Dict[str, Any]]]:
return self._request("DELETE", path, json=body, headers=headers)
# Import here to avoid circular dependency during type checking
from .emails import Emails # noqa: E402 pylint: disable=wrong-import-position
from .contacts import Contacts # noqa: E402 pylint: disable=wrong-import-position
from .contact_books import ContactBooks # noqa: E402 pylint: disable=wrong-import-position
from .domains import Domains # type: ignore # noqa: E402
from .campaigns import Campaigns # type: ignore # noqa: E402
from .webhooks import Webhooks # noqa: E402 pylint: disable=wrong-import-position