forked from digitalocean/gradient-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy path_exceptions.py
More file actions
154 lines (100 loc) · 4.38 KB
/
_exceptions.py
File metadata and controls
154 lines (100 loc) · 4.38 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
# File generated from our OpenAPI spec by Stainless. See CONTRIBUTING.md for details.
from __future__ import annotations
from typing_extensions import Literal
import httpx
__all__ = [
"BadRequestError",
"AuthenticationError",
"PermissionDeniedError",
"NotFoundError",
"ConflictError",
"UnprocessableEntityError",
"RateLimitError",
"InternalServerError",
"IndexingJobError",
"IndexingJobTimeoutError",
"AgentDeploymentError",
"AgentDeploymentTimeoutError",
]
class GradientError(Exception):
pass
class APIError(GradientError):
message: str
request: httpx.Request
body: object | None
"""The API response body.
If the API responded with a valid JSON structure then this property will be the
decoded result.
If it isn't a valid JSON structure then this will be the raw response.
If there was no response associated with this error then it will be `None`.
"""
def __init__(self, message: str, request: httpx.Request, *, body: object | None) -> None: # noqa: ARG002
super().__init__(message)
self.request = request
self.message = message
self.body = body
class APIResponseValidationError(APIError):
response: httpx.Response
status_code: int
def __init__(self, response: httpx.Response, body: object | None, *, message: str | None = None) -> None:
super().__init__(message or "Data returned by API invalid for expected schema.", response.request, body=body)
self.response = response
self.status_code = response.status_code
class APIStatusError(APIError):
"""Raised when an API response has a status code of 4xx or 5xx."""
response: httpx.Response
status_code: int
def __init__(self, message: str, *, response: httpx.Response, body: object | None) -> None:
super().__init__(message, response.request, body=body)
self.response = response
self.status_code = response.status_code
class APIConnectionError(APIError):
def __init__(self, *, message: str = "Connection error.", request: httpx.Request) -> None:
super().__init__(message, request, body=None)
class APITimeoutError(APIConnectionError):
def __init__(self, request: httpx.Request) -> None:
super().__init__(message="Request timed out.", request=request)
class BadRequestError(APIStatusError):
status_code: Literal[400] = 400 # pyright: ignore[reportIncompatibleVariableOverride]
class AuthenticationError(APIStatusError):
status_code: Literal[401] = 401 # pyright: ignore[reportIncompatibleVariableOverride]
class PermissionDeniedError(APIStatusError):
status_code: Literal[403] = 403 # pyright: ignore[reportIncompatibleVariableOverride]
class NotFoundError(APIStatusError):
status_code: Literal[404] = 404 # pyright: ignore[reportIncompatibleVariableOverride]
class ConflictError(APIStatusError):
status_code: Literal[409] = 409 # pyright: ignore[reportIncompatibleVariableOverride]
class UnprocessableEntityError(APIStatusError):
status_code: Literal[422] = 422 # pyright: ignore[reportIncompatibleVariableOverride]
class RateLimitError(APIStatusError):
status_code: Literal[429] = 429 # pyright: ignore[reportIncompatibleVariableOverride]
class InternalServerError(APIStatusError):
pass
class IndexingJobError(GradientError):
"""Raised when an indexing job fails, encounters an error, or is cancelled."""
uuid: str
phase: str
def __init__(self, message: str, *, uuid: str, phase: str) -> None:
super().__init__(message)
self.uuid = uuid
self.phase = phase
class IndexingJobTimeoutError(GradientError):
"""Raised when polling for an indexing job times out."""
uuid: str
phase: str
timeout: float
def __init__(self, message: str, *, uuid: str, phase: str, timeout: float) -> None:
super().__init__(message)
self.uuid = uuid
self.phase = phase
self.timeout = timeout
class AgentDeploymentError(GradientError):
"""Raised when an agent deployment fails."""
def __init__(self, message: str, status: str) -> None:
super().__init__(message)
self.status = status
class AgentDeploymentTimeoutError(GradientError):
"""Raised when waiting for an agent deployment times out."""
def __init__(self, message: str, agent_id: str) -> None:
super().__init__(message)
self.agent_id = agent_id