0
# Exception Handling
1

2
Comprehensive exception hierarchy for precise error handling across different cloud providers and operation types. CloudPathLib provides specific exceptions for common cloud storage scenarios, enabling robust error handling and debugging.
3

4
## Capabilities
5

6
### Exception Hierarchy
7

8
CloudPathLib defines a comprehensive exception hierarchy for precise error handling.
9

10
```python { .api }
11
class CloudPathException(Exception):
12
    """Base exception for all CloudPathLib errors."""
13

14
class AnyPathTypeError(CloudPathException):
15
    """Raised when AnyPath receives invalid input type."""
16

17
class ClientMismatchError(CloudPathException):
18
    """Raised when wrong client type is used for a path."""
19

20
class CloudPathFileExistsError(CloudPathException, FileExistsError):
21
    """Raised when attempting to create a file that already exists."""
22

23
class CloudPathFileNotFoundError(CloudPathException, FileNotFoundError):
24
    """Raised when attempting to access a file that does not exist."""
25

26
class CloudPathIsADirectoryError(CloudPathException, IsADirectoryError):
27
    """Raised when file operation is attempted on a directory."""
28

29
class CloudPathNotADirectoryError(CloudPathException, NotADirectoryError):
30
    """Raised when directory operation is attempted on a file."""
31

32
class CloudPathNotExistsError(CloudPathException):
33
    """Raised when attempting to access a path that does not exist."""
34

35
class CloudPathNotImplementedError(CloudPathException, NotImplementedError):
36
    """Raised when an operation is not supported by the cloud provider."""
37

38
class DirectoryNotEmptyError(CloudPathException):
39
    """Raised when attempting to remove a non-empty directory."""
40

41
class IncompleteImplementationError(CloudPathException):
42
    """Raised when cloud implementation is missing required components."""
43

44
class InvalidPrefixError(CloudPathException):
45
    """Raised when an invalid cloud path prefix is used."""
46

47
class InvalidConfigurationException(CloudPathException):
48
    """Raised when client configuration is invalid."""
49

50
class MissingCredentialsError(CloudPathException):
51
    """Raised when required authentication credentials are missing."""
52

53
class MissingDependenciesError(CloudPathException):
54
    """Raised when required cloud provider dependencies are not installed."""
55

56
class NoStatError(CloudPathException):
57
    """Raised when file statistics cannot be retrieved."""
58

59
class OverwriteDirtyFileError(CloudPathException):
60
    """Raised when local cached file has been modified and conflicts with cloud version."""
61

62
class OverwriteNewerCloudError(CloudPathException):
63
    """Raised when cloud file is newer than local cached version."""
64

65
class OverwriteNewerLocalError(CloudPathException):
66
    """Raised when local cached file is newer than cloud version."""
67

68
class InvalidGlobArgumentsError(CloudPathException):
69
    """Raised when invalid arguments are passed to glob operations."""
70
```
71

72
## Usage Examples
73

74
### Basic Exception Handling
75

76
```python
77
from cloudpathlib import (
78
    CloudPath,
79
    CloudPathFileNotFoundError,
80
    MissingCredentialsError,
81
    InvalidPrefixError
82
)
83

84
def safe_read_file(path_str):
85
    """Safely read a cloud file with error handling."""
86
    try:
87
        path = CloudPath(path_str)
88
        return path.read_text()
89
    
90
    except CloudPathFileNotFoundError:
91
        print(f"File not found: {path_str}")
92
        return None
93
    
94
    except MissingCredentialsError:
95
        print("Authentication credentials not configured")
96
        return None
97
    
98
    except InvalidPrefixError:
99
        print(f"Invalid cloud path prefix: {path_str}")
100
        return None
101
    
102
    except CloudPathException as e:
103
        print(f"CloudPath error: {e}")
104
        return None
105

106
# Usage
107
content = safe_read_file("s3://my-bucket/file.txt")
108
if content:
109
    print("File read successfully")
110
```
111

112
### Provider-Specific Error Handling
113

114
```python
115
from cloudpathlib import CloudPath
116
import boto3.exceptions
117
import google.api_core.exceptions
118
import azure.core.exceptions
119

120
def robust_cloud_operation(path_str):
121
    """Handle errors from different cloud providers."""
122
    try:
123
        path = CloudPath(path_str)
124
        return path.read_text()
125
    
126
    # CloudPathLib-specific exceptions
127
    except CloudPathFileNotFoundError:
128
        print("File not found in cloud storage")
129
    except MissingCredentialsError:
130
        print("Cloud credentials not configured")
131
    
132
    # AWS-specific exceptions
133
    except boto3.exceptions.Boto3Error as e:
134
        print(f"AWS error: {e}")
135
    except boto3.exceptions.NoCredentialsError:
136
        print("AWS credentials not found")
137
    
138
    # Google Cloud-specific exceptions
139
    except google.api_core.exceptions.GoogleAPIError as e:
140
        print(f"Google Cloud error: {e}")
141
    except google.api_core.exceptions.PermissionDenied:
142
        print("Google Cloud permission denied")
143
    
144
    # Azure-specific exceptions
145
    except azure.core.exceptions.AzureError as e:
146
        print(f"Azure error: {e}")
147
    except azure.core.exceptions.ClientAuthenticationError:
148
        print("Azure authentication failed")
149
    
150
    # Generic exceptions
151
    except PermissionError:
152
        print("Permission denied")
153
    except Exception as e:
154
        print(f"Unexpected error: {e}")
155
    
156
    return None
157
```
158

159
### File Operation Error Handling
160

161
```python
162
from cloudpathlib import (
163
    CloudPath,
164
    CloudPathFileExistsError,
165
    CloudPathIsADirectoryError,
166
    DirectoryNotEmptyError
167
)
168

169
def safe_file_operations():
170
    """Demonstrate error handling for file operations."""
171
    
172
    # Safe file creation
173
    def create_file_safe(path_str, content):
174
        try:
175
            path = CloudPath(path_str)
176
            
177
            # Check if file already exists
178
            if path.exists():
179
                response = input(f"File {path} exists. Overwrite? (y/n): ")
180
                if response.lower() != 'y':
181
                    return False
182
            
183
            path.write_text(content)
184
            return True
185
            
186
        except CloudPathIsADirectoryError:
187
            print(f"Error: {path_str} is a directory, not a file")
188
            return False
189
        except PermissionError:
190
            print(f"Error: No permission to write to {path_str}")
191
            return False
192
    
193
    # Safe directory removal
194
    def remove_directory_safe(path_str):
195
        try:
196
            path = CloudPath(path_str)
197
            path.rmdir()
198
            return True
199
            
200
        except CloudPathNotADirectoryError:
201
            print(f"Error: {path_str} is not a directory")
202
            return False
203
        except DirectoryNotEmptyError:
204
            print(f"Error: Directory {path_str} is not empty")
205
            # Offer to remove recursively
206
            response = input("Remove recursively? (y/n): ")
207
            if response.lower() == 'y':
208
                path.rmtree()
209
                return True
210
            return False
211
        except CloudPathFileNotFoundError:
212
            print(f"Directory {path_str} does not exist")
213
            return False
214
    
215
    # Usage examples
216
    create_file_safe("s3://bucket/file.txt", "Hello, world!")
217
    remove_directory_safe("s3://bucket/empty-dir/")
218

219
safe_file_operations()
220
```
221

222
### Retry Logic with Exception Handling
223

224
```python
225
import time
226
import random
227
from cloudpathlib import CloudPath, CloudPathException
228

229
def retry_cloud_operation(func, max_retries=3, backoff_factor=1.0):
230
    """Retry cloud operations with exponential backoff."""
231
    
232
    for attempt in range(max_retries):
233
        try:
234
            return func()
235
        
236
        except (CloudPathFileNotFoundError, CloudPathNotExistsError):
237
            # Don't retry for missing files
238
            raise
239
        
240
        except MissingCredentialsError:
241
            # Don't retry for credential issues
242
            raise
243
        
244
        except CloudPathException as e:
245
            if attempt == max_retries - 1:
246
                # Last attempt, re-raise the exception
247
                raise
248
            
249
            # Calculate backoff time
250
            backoff_time = backoff_factor * (2 ** attempt) + random.uniform(0, 1)
251
            print(f"Attempt {attempt + 1} failed: {e}")
252
            print(f"Retrying in {backoff_time:.2f} seconds...")
253
            time.sleep(backoff_time)
254
        
255
        except Exception as e:
256
            # For non-CloudPath exceptions, only retry once
257
            if attempt == 0:
258
                print(f"Unexpected error: {e}, retrying once...")
259
                time.sleep(1)
260
            else:
261
                raise
262

263
# Usage
264
def upload_file_with_retry():
265
    def upload_operation():
266
        path = CloudPath("s3://unreliable-bucket/file.txt")
267
        path.write_text("Important data")
268
        return path
269
    
270
    try:
271
        result = retry_cloud_operation(upload_operation, max_retries=3)
272
        print(f"Upload successful: {result}")
273
    except Exception as e:
274
        print(f"Upload failed after retries: {e}")
275

276
upload_file_with_retry()
277
```
278

279
### Configuration Error Handling
280

281
```python
282
from cloudpathlib import (
283
    S3Client,
284
    GSClient,
285
    AzureBlobClient,
286
    InvalidConfigurationException,
287
    MissingDependenciesError
288
)
289

290
def configure_client_safe(provider, **config):
291
    """Safely configure cloud client with error handling."""
292
    
293
    try:
294
        if provider == "s3":
295
            return S3Client(**config)
296
        elif provider == "gs":
297
            return GSClient(**config)
298
        elif provider == "azure":
299
            return AzureBlobClient(**config)
300
        else:
301
            raise ValueError(f"Unknown provider: {provider}")
302
    
303
    except MissingDependenciesError as e:
304
        print(f"Missing dependencies for {provider}: {e}")
305
        print(f"Install with: pip install cloudpathlib[{provider}]")
306
        return None
307
    
308
    except InvalidConfigurationException as e:
309
        print(f"Invalid configuration for {provider}: {e}")
310
        return None
311
    
312
    except Exception as e:
313
        print(f"Unexpected error configuring {provider} client: {e}")
314
        return None
315

316
# Usage
317
configs = {
318
    "s3": {"aws_access_key_id": "key", "aws_secret_access_key": "secret"},
319
    "gs": {"application_credentials": "path/to/creds.json"},
320
    "azure": {"connection_string": "connection_string"}
321
}
322

323
clients = {}
324
for provider, config in configs.items():
325
    client = configure_client_safe(provider, **config)
326
    if client:
327
        clients[provider] = client
328
        print(f"{provider} client configured successfully")
329
```
330

331
### Batch Operation Error Handling
332

333
```python
334
from cloudpathlib import CloudPath, CloudPathException
335
from concurrent.futures import ThreadPoolExecutor, as_completed
336

337
def process_files_batch(file_paths, process_func):
338
    """Process multiple files with error handling."""
339
    
340
    results = []
341
    errors = []
342
    
343
    def safe_process(path_str):
344
        try:
345
            path = CloudPath(path_str)
346
            result = process_func(path)
347
            return {"path": path_str, "result": result, "success": True}
348
        
349
        except CloudPathFileNotFoundError:
350
            return {"path": path_str, "error": "File not found", "success": False}
351
        except PermissionError:
352
            return {"path": path_str, "error": "Permission denied", "success": False}
353
        except CloudPathException as e:
354
            return {"path": path_str, "error": str(e), "success": False}
355
        except Exception as e:
356
            return {"path": path_str, "error": f"Unexpected: {e}", "success": False}
357
    
358
    # Process files concurrently
359
    with ThreadPoolExecutor(max_workers=5) as executor:
360
        future_to_path = {
361
            executor.submit(safe_process, path): path 
362
            for path in file_paths
363
        }
364
        
365
        for future in as_completed(future_to_path):
366
            result = future.result()
367
            
368
            if result["success"]:
369
                results.append(result)
370
            else:
371
                errors.append(result)
372
                print(f"Error processing {result['path']}: {result['error']}")
373
    
374
    return results, errors
375

376
# Usage
377
def read_file_size(path):
378
    """Example processing function."""
379
    return path.stat().st_size
380

381
file_list = [
382
    "s3://bucket/file1.txt",
383
    "s3://bucket/file2.txt",
384
    "s3://bucket/nonexistent.txt",
385
    "gs://bucket/file3.txt"
386
]
387

388
successful_results, failed_results = process_files_batch(file_list, read_file_size)
389

390
print(f"Successfully processed {len(successful_results)} files")
391
print(f"Failed to process {len(failed_results)} files")
392
```
393

394
### Custom Exception Handling
395

396
```python
397
from cloudpathlib import CloudPathException
398

399
class DataProcessingError(CloudPathException):
400
    """Custom exception for data processing errors."""
401
    pass
402

403
class InvalidDataFormatError(DataProcessingError):
404
    """Raised when data format is invalid."""
405
    pass
406

407
def process_data_file(path_str):
408
    """Process data file with custom exception handling."""
409
    
410
    try:
411
        path = CloudPath(path_str)
412
        
413
        if not path.exists():
414
            raise CloudPathFileNotFoundError(f"Data file not found: {path_str}")
415
        
416
        content = path.read_text()
417
        
418
        # Custom validation
419
        if not content.strip():
420
            raise InvalidDataFormatError(f"Data file is empty: {path_str}")
421
        
422
        if not content.startswith("DATA_VERSION"):
423
            raise InvalidDataFormatError(f"Invalid data format: {path_str}")
424
        
425
        # Process data...
426
        processed = content.upper()
427
        
428
        output_path = path.with_stem(path.stem + "_processed")
429
        output_path.write_text(processed)
430
        
431
        return output_path
432
    
433
    except InvalidDataFormatError as e:
434
        print(f"Data format error: {e}")
435
        raise  # Re-raise custom exception
436
    
437
    except DataProcessingError as e:
438
        print(f"Data processing error: {e}")
439
        raise
440
    
441
    except CloudPathException as e:
442
        print(f"Cloud storage error: {e}")
443
        raise DataProcessingError(f"Failed to process {path_str}") from e
444

445
# Usage with custom exception handling
446
try:
447
    result = process_data_file("s3://data-bucket/dataset.txt")
448
    print(f"Processing complete: {result}")
449

450
except InvalidDataFormatError:
451
    print("Fix data format and try again")
452
except DataProcessingError:
453
    print("Data processing failed")
454
except Exception as e:
455
    print(f"Unexpected error: {e}")
456
```
457

458
### Logging and Monitoring
459

460
```python
461
import logging
462
from cloudpathlib import CloudPath, CloudPathException
463

464
# Configure logging
465
logging.basicConfig(level=logging.INFO)
466
logger = logging.getLogger(__name__)
467

468
def monitored_cloud_operation(path_str, operation):
469
    """Perform cloud operation with comprehensive logging."""
470
    
471
    logger.info(f"Starting {operation} on {path_str}")
472
    
473
    try:
474
        path = CloudPath(path_str)
475
        
476
        if operation == "read":
477
            result = path.read_text()
478
            logger.info(f"Successfully read {len(result)} characters from {path_str}")
479
            return result
480
        
481
        elif operation == "exists":
482
            result = path.exists()
483
            logger.info(f"Existence check for {path_str}: {result}")
484
            return result
485
        
486
        elif operation == "stat":
487
            result = path.stat()
488
            logger.info(f"File stats for {path_str}: size={result.st_size}")
489
            return result
490
        
491
        else:
492
            raise ValueError(f"Unknown operation: {operation}")
493
    
494
    except CloudPathFileNotFoundError as e:
495
        logger.warning(f"File not found during {operation}: {path_str}")
496
        raise
497
    
498
    except MissingCredentialsError as e:
499
        logger.error(f"Credentials missing for {operation} on {path_str}")
500
        raise
501
    
502
    except CloudPathException as e:
503
        logger.error(f"CloudPath error during {operation} on {path_str}: {e}")
504
        raise
505
    
506
    except Exception as e:
507
        logger.error(f"Unexpected error during {operation} on {path_str}: {e}")
508
        raise
509

510
# Usage with monitoring
511
try:
512
    content = monitored_cloud_operation("s3://bucket/file.txt", "read")
513
    stats = monitored_cloud_operation("s3://bucket/file.txt", "stat")
514
    
515
except CloudPathException as e:
516
    logger.error(f"Operation failed: {e}")
517
```
518

519
### Error Recovery Patterns
520

521
```python
522
from cloudpathlib import CloudPath, CloudPathFileNotFoundError
523

524
def resilient_data_access(primary_path, backup_paths=None):
525
    """Access data with fallback to backup locations."""
526
    
527
    backup_paths = backup_paths or []
528
    all_paths = [primary_path] + backup_paths
529
    
530
    last_exception = None
531
    
532
    for i, path_str in enumerate(all_paths):
533
        try:
534
            path = CloudPath(path_str)
535
            content = path.read_text()
536
            
537
            if i > 0:
538
                print(f"Successfully accessed backup path: {path_str}")
539
            
540
            return content
541
        
542
        except CloudPathFileNotFoundError as e:
543
            last_exception = e
544
            if i < len(all_paths) - 1:
545
                print(f"Primary path failed, trying backup: {all_paths[i + 1]}")
546
            continue
547
        
548
        except CloudPathException as e:
549
            last_exception = e
550
            print(f"Error accessing {path_str}: {e}")
551
            continue
552
    
553
    # All paths failed
554
    raise CloudPathFileNotFoundError(
555
        f"Could not access data from any location. Last error: {last_exception}"
556
    )
557

558
# Usage
559
try:
560
    data = resilient_data_access(
561
        primary_path="s3://primary-bucket/data.txt",
562
        backup_paths=[
563
            "s3://backup-bucket/data.txt",
564
            "gs://backup-bucket/data.txt",
565
            "/local/backup/data.txt"
566
        ]
567
    )
568
    print("Data accessed successfully")
569
    
570
except CloudPathException as e:
571
    print(f"All data sources failed: {e}")
572
```