0
# Cloud-Specific Operations
1
2
Advanced cloud storage features that go beyond traditional filesystem operations. These capabilities leverage cloud-native features like URL generation, presigned URLs, efficient copying, upload/download operations, caching management, and cloud service metadata access.
3
4
## Capabilities
5
6
### File Upload and Download
7
8
Transfer files between local filesystem and cloud storage.
9
10
```python { .api }
11
def download_to(
12
self,
13
destination: typing.Union[str, "os.PathLike"]
14
) -> "pathlib.Path":
15
"""
16
Download cloud file to local destination.
17
18
Args:
19
destination: Local path for downloaded file
20
21
Returns:
22
pathlib.Path object for downloaded file
23
"""
24
25
def upload_from(
26
self,
27
source: typing.Union[str, "os.PathLike"],
28
**kwargs
29
) -> "CloudPath":
30
"""
31
Upload local file to cloud storage.
32
33
Args:
34
source: Local file path to upload
35
**kwargs: Cloud-specific upload options
36
37
Returns:
38
CloudPath object for uploaded file
39
"""
40
```
41
42
### File and Directory Copying
43
44
Copy files and directories within and between cloud storage locations.
45
46
```python { .api }
47
def copy(
48
self,
49
destination: typing.Union[str, "CloudPath"],
50
**kwargs
51
) -> "CloudPath":
52
"""
53
Copy file to destination.
54
55
Args:
56
destination: Target path for copy
57
**kwargs: Cloud-specific copy options
58
59
Returns:
60
CloudPath object for copied file
61
"""
62
63
def copytree(
64
self,
65
destination: typing.Union[str, "CloudPath"],
66
**kwargs
67
) -> "CloudPath":
68
"""
69
Copy directory tree to destination.
70
71
Args:
72
destination: Target path for directory copy
73
**kwargs: Cloud-specific copy options
74
75
Returns:
76
CloudPath object for copied directory
77
"""
78
```
79
80
### URL Generation
81
82
Generate URLs for cloud resources with optional presigning.
83
84
```python { .api }
85
def as_url(
86
self,
87
presign: bool = False,
88
expire_seconds: int = 3600
89
) -> str:
90
"""
91
Get URL for cloud resource.
92
93
Args:
94
presign: Generate presigned URL for temporary access
95
expire_seconds: Expiration time for presigned URLs
96
97
Returns:
98
URL string for the cloud resource
99
"""
100
```
101
102
### Cache Management
103
104
Control local file caching behavior.
105
106
```python { .api }
107
def clear_cache(self) -> None:
108
"""
109
Clear local cache for this path.
110
Removes any locally cached copy of the file.
111
"""
112
113
@property
114
def fspath(self) -> str:
115
"""
116
Get local filesystem path for cached file.
117
Downloads and caches file if not already cached.
118
119
Returns:
120
Local path to cached file
121
"""
122
```
123
124
### File Movement and Deletion
125
126
Move and delete cloud files and directories.
127
128
```python { .api }
129
def unlink(self, missing_ok: bool = False) -> None:
130
"""
131
Delete file from cloud storage.
132
133
Args:
134
missing_ok: Don't raise error if file doesn't exist
135
136
Raises:
137
CloudPathFileNotFoundError: File not found and missing_ok=False
138
"""
139
140
def rename(self, target: typing.Union[str, "CloudPath"]) -> "CloudPath":
141
"""
142
Rename/move file to target location.
143
144
Args:
145
target: New path for the file
146
147
Returns:
148
CloudPath object for renamed file
149
"""
150
151
def replace(self, target: typing.Union[str, "CloudPath"]) -> "CloudPath":
152
"""
153
Replace target with this file.
154
155
Args:
156
target: Path to replace
157
158
Returns:
159
CloudPath object for replacement location
160
"""
161
```
162
163
## Usage Examples
164
165
### File Upload and Download
166
167
```python
168
from cloudpathlib import CloudPath
169
import pathlib
170
171
# Upload local file to cloud
172
local_file = pathlib.Path("local_data.csv")
173
cloud_path = CloudPath("s3://my-bucket/data.csv")
174
175
# Upload file
176
cloud_path.upload_from(local_file)
177
print(f"Uploaded to {cloud_path}")
178
179
# Download file from cloud
180
downloaded_path = cloud_path.download_to("downloaded_data.csv")
181
print(f"Downloaded to {downloaded_path}")
182
```
183
184
### Directory Upload and Download
185
186
```python
187
# Upload entire directory
188
local_dir = pathlib.Path("local_project/")
189
cloud_dir = CloudPath("s3://my-bucket/project/")
190
191
# Upload directory tree
192
for local_path in local_dir.rglob("*"):
193
if local_path.is_file():
194
relative_path = local_path.relative_to(local_dir)
195
cloud_file = cloud_dir / relative_path
196
cloud_file.upload_from(local_path)
197
198
# Download directory tree
199
for cloud_file in cloud_dir.rglob("*"):
200
if cloud_file.is_file():
201
relative_path = cloud_file.relative_to(cloud_dir)
202
local_path = pathlib.Path("downloads") / relative_path
203
local_path.parent.mkdir(parents=True, exist_ok=True)
204
cloud_file.download_to(local_path)
205
```
206
207
### Cloud-to-Cloud Copying
208
209
```python
210
# Copy between cloud providers
211
s3_path = CloudPath("s3://source-bucket/file.txt")
212
gs_path = CloudPath("gs://dest-bucket/file.txt")
213
214
# Copy file between clouds (via local cache)
215
s3_path.copy(gs_path)
216
217
# Copy within same cloud service
218
source = CloudPath("s3://my-bucket/source/file.txt")
219
destination = CloudPath("s3://my-bucket/backup/file.txt")
220
source.copy(destination)
221
```
222
223
### Directory Copying
224
225
```python
226
# Copy entire directory tree
227
source_dir = CloudPath("s3://source-bucket/data/")
228
dest_dir = CloudPath("s3://backup-bucket/data-backup/")
229
230
# Copy directory tree
231
source_dir.copytree(dest_dir)
232
233
# Copy with filtering
234
source_dir = CloudPath("s3://my-bucket/project/")
235
backup_dir = CloudPath("s3://backup-bucket/project-backup/")
236
237
# Only copy specific file types
238
for file_path in source_dir.rglob("*"):
239
if file_path.suffix in ['.py', '.md', '.txt']:
240
relative_path = file_path.relative_to(source_dir)
241
dest_path = backup_dir / relative_path
242
file_path.copy(dest_path)
243
```
244
245
### URL Generation
246
247
```python
248
# Get public URL
249
cloud_path = CloudPath("s3://public-bucket/image.jpg")
250
public_url = cloud_path.as_url()
251
print(f"Public URL: {public_url}")
252
253
# Generate presigned URL for temporary access
254
private_path = CloudPath("s3://private-bucket/document.pdf")
255
signed_url = private_path.as_url(presign=True, expire_seconds=3600)
256
print(f"Temporary URL (1 hour): {signed_url}")
257
258
# Short-term presigned URL
259
temp_url = private_path.as_url(presign=True, expire_seconds=300) # 5 minutes
260
print(f"Short-term URL: {temp_url}")
261
```
262
263
### Cache Management
264
265
```python
266
# Work with cached files
267
cloud_path = CloudPath("s3://my-bucket/large-file.dat")
268
269
# Get local cached path (downloads if needed)
270
local_cache_path = cloud_path.fspath
271
print(f"Cached at: {local_cache_path}")
272
273
# Work with local copy
274
with open(local_cache_path, 'rb') as f:
275
data = f.read(1024)
276
277
# Clear cache when done
278
cloud_path.clear_cache()
279
280
# Force re-download on next access
281
fresh_cache_path = cloud_path.fspath
282
```
283
284
### File Movement and Deletion
285
286
```python
287
# Move/rename files
288
old_path = CloudPath("s3://my-bucket/old-name.txt")
289
new_path = CloudPath("s3://my-bucket/new-name.txt")
290
291
# Rename file
292
renamed = old_path.rename(new_path)
293
print(f"Renamed to: {renamed}")
294
295
# Move to different directory
296
source = CloudPath("s3://my-bucket/temp/file.txt")
297
target = CloudPath("s3://my-bucket/archive/file.txt")
298
moved = source.replace(target)
299
300
# Delete files
301
file_to_delete = CloudPath("s3://my-bucket/unwanted.txt")
302
file_to_delete.unlink()
303
304
# Safe deletion
305
try:
306
file_to_delete.unlink()
307
except CloudPathFileNotFoundError:
308
print("File already deleted")
309
310
# Safe deletion with missing_ok
311
file_to_delete.unlink(missing_ok=True)
312
```
313
314
### Batch Operations
315
316
```python
317
# Batch upload
318
local_files = pathlib.Path("local_data/").glob("*.csv")
319
cloud_base = CloudPath("s3://my-bucket/csv-data/")
320
321
for local_file in local_files:
322
cloud_file = cloud_base / local_file.name
323
cloud_file.upload_from(local_file)
324
print(f"Uploaded {local_file.name}")
325
326
# Batch download
327
cloud_files = CloudPath("s3://my-bucket/results/").glob("*.json")
328
local_base = pathlib.Path("downloads/")
329
local_base.mkdir(exist_ok=True)
330
331
for cloud_file in cloud_files:
332
local_file = local_base / cloud_file.name
333
cloud_file.download_to(local_file)
334
print(f"Downloaded {cloud_file.name}")
335
```
336
337
### Monitoring Transfer Progress
338
339
```python
340
def upload_with_progress(local_path, cloud_path):
341
"""Upload with progress monitoring."""
342
file_size = local_path.stat().st_size
343
print(f"Uploading {local_path.name} ({file_size} bytes)")
344
345
# Upload file
346
result = cloud_path.upload_from(local_path)
347
print(f"Upload complete: {result}")
348
return result
349
350
def download_with_progress(cloud_path, local_path):
351
"""Download with progress monitoring."""
352
print(f"Downloading {cloud_path.name}")
353
354
# Download file
355
result = cloud_path.download_to(local_path)
356
downloaded_size = result.stat().st_size
357
print(f"Downloaded {downloaded_size} bytes to {result}")
358
return result
359
360
# Usage
361
local_file = pathlib.Path("data.csv")
362
cloud_file = CloudPath("s3://my-bucket/data.csv")
363
364
upload_with_progress(local_file, cloud_file)
365
download_with_progress(cloud_file, "downloaded.csv")
366
```
367
368
### Working with Temporary URLs
369
370
```python
371
import requests
372
373
# Generate temporary download URL
374
cloud_path = CloudPath("s3://private-bucket/report.pdf")
375
download_url = cloud_path.as_url(presign=True, expire_seconds=1800) # 30 minutes
376
377
# Use URL with external tools
378
response = requests.get(download_url)
379
with open("downloaded_report.pdf", "wb") as f:
380
f.write(response.content)
381
382
# Generate upload URL (for some cloud providers)
383
upload_path = CloudPath("s3://uploads-bucket/new-file.txt")
384
upload_url = upload_path.as_url(presign=True, expire_seconds=600) # 10 minutes
385
386
# External service can upload directly to this URL
387
print(f"Upload URL: {upload_url}")
388
```
389
390
### Advanced Copy Operations
391
392
```python
393
# Copy with metadata preservation
394
source = CloudPath("s3://source/important.doc")
395
destination = CloudPath("s3://backup/important.doc")
396
397
# Copy with cloud-specific options (varies by provider)
398
source.copy(destination, preserve_metadata=True)
399
400
# Conditional copying based on modification time
401
if not destination.exists() or source.stat().st_mtime > destination.stat().st_mtime:
402
print("Source is newer, copying...")
403
source.copy(destination)
404
else:
405
print("Destination is up to date")
406
407
# Copy with different storage class (cloud-specific)
408
source.copy(destination, storage_class='GLACIER') # S3 example
409
```