-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy patherrors.py
More file actions
193 lines (154 loc) · 5.14 KB
/
Copy patherrors.py
File metadata and controls
193 lines (154 loc) · 5.14 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
"""Exception hierarchy for the cloudlayer.io Python SDK.
Mirrors the JS SDK's 8 error classes with request context for debugging.
"""
from __future__ import annotations
from typing import Any
class CloudLayerError(Exception):
"""Base error class for all CloudLayer SDK errors."""
def __init__(self, message: str) -> None:
self.message = message
super().__init__(message)
def __repr__(self) -> str:
return f"{self.__class__.__name__}({self.message!r})"
class CloudLayerConfigError(CloudLayerError):
"""Thrown when the CloudLayer client is configured with invalid options."""
def __init__(self, message: str) -> None:
super().__init__(message)
class CloudLayerValidationError(CloudLayerError):
"""Thrown when client-side input validation fails before making the API request."""
field: str
def __init__(self, field: str, message: str) -> None:
self.field = field
super().__init__(f'Validation error on "{field}": {message}')
def __repr__(self) -> str:
return f"{self.__class__.__name__}(field={self.field!r}, message={self.message!r})"
class CloudLayerNetworkError(CloudLayerError):
"""Thrown when a network error occurs (DNS failure, connection refused, etc.)."""
request_path: str
request_method: str
def __init__(
self,
message: str,
*,
request_path: str,
request_method: str,
) -> None:
self.request_path = request_path
self.request_method = request_method
super().__init__(message)
def __repr__(self) -> str:
return (
f"{self.__class__.__name__}("
f"message={self.message!r}, "
f"request_path={self.request_path!r}, "
f"request_method={self.request_method!r})"
)
class CloudLayerTimeoutError(CloudLayerError):
"""Thrown when a request times out."""
timeout: int
request_path: str
request_method: str
def __init__(
self,
*,
timeout: int,
request_path: str,
request_method: str,
) -> None:
self.timeout = timeout
self.request_path = request_path
self.request_method = request_method
super().__init__(f"Request timed out after {timeout}ms: {request_method} {request_path}")
def __repr__(self) -> str:
return (
f"{self.__class__.__name__}("
f"timeout={self.timeout!r}, "
f"request_path={self.request_path!r}, "
f"request_method={self.request_method!r})"
)
class CloudLayerApiError(CloudLayerError):
"""Thrown when the CloudLayer API returns an error response (4xx/5xx)."""
status: int
status_text: str
body: Any
request_path: str
request_method: str
def __init__(
self,
message: str,
*,
status: int,
status_text: str,
body: Any = None,
request_path: str,
request_method: str,
) -> None:
self.status = status
self.status_text = status_text
self.body = body
self.request_path = request_path
self.request_method = request_method
super().__init__(message)
def __repr__(self) -> str:
return (
f"{self.__class__.__name__}("
f"status={self.status!r}, "
f"request_path={self.request_path!r}, "
f"request_method={self.request_method!r})"
)
class CloudLayerAuthError(CloudLayerApiError):
"""Thrown when the API rejects the request due to invalid or missing authentication.
HTTP 401 (Unauthorized) or 403 (Forbidden).
"""
def __init__(
self,
*,
status: int,
status_text: str,
body: Any = None,
request_path: str,
request_method: str,
) -> None:
message = (
"Authentication failed: invalid or missing API key"
if status == 401
else "Authorization failed: insufficient permissions"
)
super().__init__(
message,
status=status,
status_text=status_text,
body=body,
request_path=request_path,
request_method=request_method,
)
class CloudLayerRateLimitError(CloudLayerApiError):
"""Thrown when the API rate limit is exceeded (HTTP 429)."""
retry_after: int | None
def __init__(
self,
*,
status: int,
status_text: str,
body: Any = None,
request_path: str,
request_method: str,
retry_after: int | None = None,
) -> None:
self.retry_after = retry_after
retry_msg = f" Retry after {retry_after}s." if retry_after else ""
super().__init__(
f"Rate limit exceeded.{retry_msg}",
status=status,
status_text=status_text,
body=body,
request_path=request_path,
request_method=request_method,
)
def __repr__(self) -> str:
return (
f"{self.__class__.__name__}("
f"status={self.status!r}, "
f"retry_after={self.retry_after!r}, "
f"request_path={self.request_path!r})"
)