0
# Exception Handling
1
2
Comprehensive exception hierarchy for handling API errors, authentication issues, and operational failures.
3
4
## Capabilities
5
6
### Base Exception Classes
7
8
Core exception classes that form the foundation of Cloudinary error handling.
9
10
```python { .api }
11
class Error(Exception):
12
"""Base exception class for all Cloudinary errors.
13
14
All Cloudinary-specific exceptions inherit from this base class, allowing
15
for comprehensive error handling and logging.
16
17
Attributes:
18
message (str): Error message describing the issue
19
http_code (int): HTTP status code associated with the error (if applicable)
20
"""
21
22
def __init__(self, message=None, http_code=None):
23
"""Initialize base error.
24
25
Args:
26
message (str, optional): Error message
27
http_code (int, optional): HTTP status code
28
"""
29
super(Error, self).__init__(message)
30
self.message = message
31
self.http_code = http_code
32
33
def __str__(self):
34
"""Return string representation of the error."""
35
return self.message or super(Error, self).__str__()
36
```
37
38
### HTTP and API Errors
39
40
Exceptions related to HTTP communication and API responses.
41
42
```python { .api }
43
class NotFound(Error):
44
"""Exception raised when a requested resource is not found.
45
46
Typically corresponds to HTTP 404 responses from the Cloudinary API.
47
Common scenarios:
48
- Requesting a non-existent public_id
49
- Accessing a deleted resource
50
- Referencing an invalid transformation
51
"""
52
53
class NotAllowed(Error):
54
"""Exception raised when an operation is not permitted.
55
56
Typically corresponds to HTTP 403 responses indicating insufficient permissions
57
or account limitations.
58
Common scenarios:
59
- Attempting operations without proper API credentials
60
- Exceeding account quotas or limits
61
- Accessing restricted functionality
62
"""
63
64
class AlreadyExists(Error):
65
"""Exception raised when attempting to create a resource that already exists.
66
67
Common scenarios:
68
- Creating a transformation with an existing name
69
- Uploading with overwrite=False to an existing public_id
70
- Creating upload presets with duplicate names
71
"""
72
73
class RateLimited(Error):
74
"""Exception raised when API rate limits are exceeded.
75
76
Indicates that too many requests have been made within the allowed time window.
77
The client should implement backoff and retry logic.
78
79
Attributes:
80
retry_after (int): Seconds to wait before retrying (if provided by API)
81
"""
82
83
def __init__(self, message=None, http_code=None, retry_after=None):
84
"""Initialize rate limit error.
85
86
Args:
87
message (str, optional): Error message
88
http_code (int, optional): HTTP status code (typically 429)
89
retry_after (int, optional): Seconds to wait before retry
90
"""
91
super(RateLimited, self).__init__(message, http_code)
92
self.retry_after = retry_after
93
94
class BadRequest(Error):
95
"""Exception raised for malformed requests or invalid parameters.
96
97
Typically corresponds to HTTP 400 responses indicating client-side errors.
98
Common scenarios:
99
- Invalid transformation parameters
100
- Malformed public_ids or resource identifiers
101
- Missing required parameters
102
- Invalid file formats or types
103
"""
104
105
class GeneralError(Error):
106
"""Exception raised for general API errors that don't fit other categories.
107
108
Used for unexpected server errors, temporary service issues, or other
109
unspecified problems with API operations.
110
"""
111
112
class AuthorizationRequired(Error):
113
"""Exception raised when authentication is required but missing or invalid.
114
115
Typically corresponds to HTTP 401 responses indicating authentication failures.
116
Common scenarios:
117
- Missing API credentials
118
- Invalid API key or secret
119
- Expired or malformed authentication tokens
120
- Attempting authenticated operations without proper credentials
121
"""
122
```
123
124
## Error Handling Patterns
125
126
### Basic Exception Handling
127
128
```python
129
from cloudinary import uploader, api
130
from cloudinary.exceptions import Error, NotFound, NotAllowed, BadRequest, AuthorizationRequired
131
132
try:
133
# Upload operation
134
result = uploader.upload("nonexistent_file.jpg")
135
except NotFound as e:
136
print(f"File not found: {e}")
137
except BadRequest as e:
138
print(f"Invalid request: {e}")
139
except AuthorizationRequired as e:
140
print(f"Authentication failed: {e}")
141
except Error as e:
142
print(f"Cloudinary error: {e}")
143
if hasattr(e, 'http_code'):
144
print(f"HTTP status: {e.http_code}")
145
except Exception as e:
146
print(f"Unexpected error: {e}")
147
```
148
149
### API Operation Error Handling
150
151
```python
152
from cloudinary import api
153
from cloudinary.exceptions import NotFound, NotAllowed, RateLimited
154
155
def safe_get_resource(public_id):
156
"""Safely retrieve resource with comprehensive error handling."""
157
try:
158
return api.resource(public_id)
159
except NotFound:
160
print(f"Resource '{public_id}' not found")
161
return None
162
except NotAllowed as e:
163
print(f"Access denied for '{public_id}': {e}")
164
return None
165
except RateLimited as e:
166
print(f"Rate limited. Retry after: {getattr(e, 'retry_after', 60)} seconds")
167
return None
168
except Error as e:
169
print(f"Cloudinary API error: {e}")
170
return None
171
172
def safe_delete_resources(public_ids):
173
"""Safely delete multiple resources with error handling."""
174
try:
175
result = api.delete_resources(public_ids)
176
177
# Check for partial failures
178
if result.get('partial'):
179
print("Some resources could not be deleted:")
180
for public_id in result.get('deleted', {}):
181
if result['deleted'][public_id] != 'deleted':
182
print(f" {public_id}: {result['deleted'][public_id]}")
183
184
return result
185
except NotAllowed as e:
186
print(f"Deletion not allowed: {e}")
187
return None
188
except Error as e:
189
print(f"Error during deletion: {e}")
190
return None
191
```
192
193
### Upload Error Handling
194
195
```python
196
from cloudinary import uploader
197
from cloudinary.exceptions import BadRequest, NotAllowed, GeneralError
198
import time
199
200
def robust_upload(file_path, **options):
201
"""Upload with retry logic and comprehensive error handling."""
202
max_retries = 3
203
retry_delay = 1
204
205
for attempt in range(max_retries):
206
try:
207
result = uploader.upload(file_path, **options)
208
print(f"Upload successful: {result['public_id']}")
209
return result
210
211
except BadRequest as e:
212
print(f"Invalid upload parameters: {e}")
213
# Don't retry for bad requests
214
return None
215
216
except NotAllowed as e:
217
print(f"Upload not allowed: {e}")
218
return None
219
220
except RateLimited as e:
221
retry_after = getattr(e, 'retry_after', retry_delay * (2 ** attempt))
222
print(f"Rate limited. Waiting {retry_after} seconds...")
223
time.sleep(retry_after)
224
continue
225
226
except GeneralError as e:
227
if attempt < max_retries - 1:
228
print(f"Upload failed (attempt {attempt + 1}): {e}")
229
print(f"Retrying in {retry_delay} seconds...")
230
time.sleep(retry_delay)
231
retry_delay *= 2
232
continue
233
else:
234
print(f"Upload failed after {max_retries} attempts: {e}")
235
return None
236
237
except Error as e:
238
print(f"Unexpected Cloudinary error: {e}")
239
return None
240
241
except Exception as e:
242
print(f"System error during upload: {e}")
243
return None
244
245
return None
246
247
# Usage
248
result = robust_upload(
249
"image.jpg",
250
public_id="my_image",
251
folder="uploads",
252
overwrite=True
253
)
254
255
if result:
256
print(f"Successfully uploaded: {result['secure_url']}")
257
else:
258
print("Upload failed after all retry attempts")
259
```
260
261
### Batch Operation Error Handling
262
263
```python
264
from cloudinary import api
265
from cloudinary.exceptions import Error
266
267
def process_resources_safely(public_ids, operation_func, **kwargs):
268
"""Process multiple resources with individual error handling."""
269
results = []
270
errors = []
271
272
for public_id in public_ids:
273
try:
274
result = operation_func(public_id, **kwargs)
275
results.append({'public_id': public_id, 'success': True, 'result': result})
276
except NotFound:
277
error_msg = f"Resource '{public_id}' not found"
278
errors.append({'public_id': public_id, 'error': error_msg})
279
results.append({'public_id': public_id, 'success': False, 'error': error_msg})
280
except NotAllowed as e:
281
error_msg = f"Operation not allowed for '{public_id}': {e}"
282
errors.append({'public_id': public_id, 'error': error_msg})
283
results.append({'public_id': public_id, 'success': False, 'error': error_msg})
284
except Error as e:
285
error_msg = f"Error processing '{public_id}': {e}"
286
errors.append({'public_id': public_id, 'error': error_msg})
287
results.append({'public_id': public_id, 'success': False, 'error': error_msg})
288
289
return {
290
'results': results,
291
'errors': errors,
292
'success_count': len([r for r in results if r['success']]),
293
'error_count': len(errors)
294
}
295
296
# Usage examples
297
def get_resource_details(public_id):
298
return api.resource(public_id)
299
300
def update_resource_tags(public_id, tags):
301
return api.update(public_id, tags=tags)
302
303
# Process multiple resources
304
public_ids = ["image1", "image2", "nonexistent", "image3"]
305
306
# Get details for all resources
307
details_result = process_resources_safely(public_ids, get_resource_details)
308
print(f"Successfully retrieved {details_result['success_count']} resources")
309
print(f"Failed to retrieve {details_result['error_count']} resources")
310
311
# Update tags for all resources
312
tag_result = process_resources_safely(
313
public_ids,
314
update_resource_tags,
315
tags=["batch_processed", "updated"]
316
)
317
```
318
319
### Advanced Error Recovery
320
321
```python
322
import logging
323
from cloudinary.exceptions import *
324
325
# Configure logging
326
logging.basicConfig(level=logging.INFO)
327
logger = logging.getLogger(__name__)
328
329
class CloudinaryErrorHandler:
330
"""Advanced error handler with logging and recovery strategies."""
331
332
def __init__(self):
333
self.error_counts = {}
334
335
def handle_error(self, operation, error, context=None):
336
"""Handle errors with logging and recovery suggestions."""
337
error_type = type(error).__name__
338
self.error_counts[error_type] = self.error_counts.get(error_type, 0) + 1
339
340
logger.error(f"{operation} failed: {error_type} - {error}")
341
342
if context:
343
logger.error(f"Context: {context}")
344
345
# Provide recovery suggestions
346
if isinstance(error, NotFound):
347
logger.info("Recovery: Check if the resource exists or verify the public_id")
348
elif isinstance(error, AuthorizationRequired):
349
logger.info("Recovery: Verify API credentials and permissions")
350
elif isinstance(error, RateLimited):
351
retry_after = getattr(error, 'retry_after', 60)
352
logger.info(f"Recovery: Wait {retry_after} seconds before retrying")
353
elif isinstance(error, BadRequest):
354
logger.info("Recovery: Check request parameters and format")
355
elif isinstance(error, NotAllowed):
356
logger.info("Recovery: Check account limits and permissions")
357
358
def get_error_summary(self):
359
"""Get summary of encountered errors."""
360
return dict(self.error_counts)
361
362
# Usage
363
error_handler = CloudinaryErrorHandler()
364
365
try:
366
result = api.resource("nonexistent_image")
367
except Error as e:
368
error_handler.handle_error(
369
"Get resource",
370
e,
371
context={"public_id": "nonexistent_image"}
372
)
373
374
# Get error statistics
375
print("Error summary:", error_handler.get_error_summary())
376
```