tessl/pypi-cloudpathlib
Pathlib-style classes for cloud storage services that provide seamless access to AWS S3, Google Cloud Storage, and Azure Blob Storage with familiar filesystem operations.
—
Pending
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
file-io.mddocs/
File I/O Operations
Comprehensive file input/output capabilities with support for text and binary modes, streaming operations, and cloud-specific optimizations. All file operations automatically handle cloud storage specifics like authentication, caching, and efficient data transfer.
Capabilities
File Opening
Open cloud files with familiar Python file I/O patterns.
def open(
self,
mode: str = "r",
buffering: int = -1,
encoding: typing.Optional[str] = None,
errors: typing.Optional[str] = None,
newline: typing.Optional[str] = None,
force_overwrite_from_cloud: typing.Optional[bool] = None,
force_overwrite_to_cloud: typing.Optional[bool] = None
) -> typing.IO:
"""
Open cloud file for reading or writing.
Args:
mode: File mode ('r', 'w', 'a', 'rb', 'wb', 'ab')
buffering: Buffer size (-1 for default)
encoding: Text encoding (for text modes)
errors: Error handling strategy
newline: Newline handling
force_overwrite_from_cloud: Force refresh from cloud storage
force_overwrite_to_cloud: Force upload to cloud storage
Returns:
File object compatible with built-in open()
"""Text File Operations
Read and write text content with automatic encoding handling.
def read_text(
self,
encoding: str = None,
errors: str = None
) -> str:
"""
Read entire file as text.
Args:
encoding: Text encoding (default: utf-8)
errors: Error handling ('strict', 'ignore', 'replace')
Returns:
File contents as string
"""
def write_text(
self,
data: str,
encoding: str = None,
errors: str = None,
newline: str = None
) -> int:
"""
Write text data to file.
Args:
data: Text content to write
encoding: Text encoding (default: utf-8)
errors: Error handling strategy
newline: Newline handling
Returns:
Number of characters written
"""Binary File Operations
Handle binary data with efficient streaming and memory management.
def read_bytes(self) -> bytes:
"""
Read entire file as bytes.
Returns:
File contents as bytes
"""
def write_bytes(self, data: bytes) -> int:
"""
Write binary data to file.
Args:
data: Binary content to write
Returns:
Number of bytes written
"""File Metadata
Access file information and properties.
def stat(self) -> "os.stat_result":
"""
Get file statistics.
Returns:
os.stat_result object with file metadata
Fields include: st_size, st_mtime, st_mode
"""
def touch(self, exist_ok: bool = True) -> None:
"""
Create file or update timestamp.
Args:
exist_ok: Don't raise error if file exists
"""Usage Examples
Basic File Reading
from cloudpathlib import CloudPath
path = CloudPath("s3://my-bucket/data.txt")
# Read entire file as text
content = path.read_text()
print(content)
# Read with specific encoding
content = path.read_text(encoding='utf-8')
# Read binary data
binary_data = path.read_bytes()Basic File Writing
path = CloudPath("s3://my-bucket/output.txt")
# Write text
path.write_text("Hello, cloud storage!")
# Write with specific encoding
path.write_text("Hello, world! 🌍", encoding='utf-8')
# Write binary data
path.write_bytes(b"Binary content")Streaming File Operations
# Stream reading for large files
path = CloudPath("s3://my-bucket/large-file.txt")
with path.open('r') as f:
for line in f:
process_line(line)
# Stream writing
path = CloudPath("s3://my-bucket/output.txt")
with path.open('w') as f:
f.write("Line 1\n")
f.write("Line 2\n")
f.flush() # Force uploadBinary File Handling
# Read binary file in chunks
path = CloudPath("s3://my-bucket/image.jpg")
with path.open('rb') as f:
while True:
chunk = f.read(8192) # 8KB chunks
if not chunk:
break
process_chunk(chunk)
# Write binary data
path = CloudPath("s3://my-bucket/data.bin")
with path.open('wb') as f:
f.write(b'\x00\x01\x02\x03')File Modes and Options
path = CloudPath("s3://my-bucket/file.txt")
# Different file modes
with path.open('r') as f: # Read text
content = f.read()
with path.open('rb') as f: # Read binary
data = f.read()
with path.open('w') as f: # Write text (overwrites)
f.write("New content")
with path.open('a') as f: # Append text
f.write("Additional content")
with path.open('wb') as f: # Write binary
f.write(b"Binary data")Advanced File Operations
# Check file properties
path = CloudPath("s3://my-bucket/file.txt")
if path.exists():
# Get file statistics
stats = path.stat()
print(f"Size: {stats.st_size} bytes")
print(f"Modified: {stats.st_mtime}")
# Create/update file timestamp
path.touch()Error Handling
from cloudpathlib import CloudPathFileNotFoundError
path = CloudPath("s3://my-bucket/nonexistent.txt")
try:
content = path.read_text()
except CloudPathFileNotFoundError:
print("File not found")
except PermissionError:
print("Access denied")
# Safe file operations
if path.exists():
content = path.read_text()
else:
print("File does not exist")Working with Different Encodings
# UTF-8 text (default)
path = CloudPath("s3://my-bucket/utf8.txt")
path.write_text("Hello 世界", encoding='utf-8')
content = path.read_text(encoding='utf-8')
# Other encodings
path = CloudPath("s3://my-bucket/latin1.txt")
path.write_text("Café", encoding='latin-1')
content = path.read_text(encoding='latin-1')
# Handle encoding errors
try:
content = path.read_text(encoding='ascii', errors='strict')
except UnicodeDecodeError:
content = path.read_text(encoding='ascii', errors='ignore')Context Manager Usage
# Automatic resource cleanup
path = CloudPath("s3://my-bucket/data.txt")
# Text operations
with path.open('w', encoding='utf-8') as f:
f.write("Content is automatically uploaded when context exits")
# Binary operations
with path.open('rb') as f:
data = f.read(1024) # Read first 1KB
# Multiple operations
with path.open('a') as f:
f.write("Line 1\n")
f.write("Line 2\n")
# File is automatically closed and uploadedInstall with Tessl CLI
npx tessl i tessl/pypi-cloudpathlib