DOCSEARCHbyAlgolia
Submit Docs

s3path

repository·master·Indexed 42 days ago

https://github.com/liormizr/s3path

A Pythonic, File-System/Path-like interface for AWS S3 that mimics the pathlib API using boto3 as the underlying driver. It provides tools for managing S3 buckets and objects, including support for globbing, recursive searches, pre-signed URLs, and S3-compatible storage like LocalStack and MinIO.

Tokens
6.3K
Snippets
27
Records
29
Agent score
52%

What's inside s3path

  1. Use PureS3Path for S3 path manipulation

    master

    PureS3Path is a subclass of PurePath designed for AWS S3 semantics. It allows for path manipulation without performing actual I/O operations.

    Key Differences from PurePosixPath:

    • S3 does not support symbolic links, so double dots ('..') are resolved by collapsing the path: PureS3Path('foo/../bar') becomes PureS3Path('bar').
    • All methods will raise ValueError if the path is not absolute.

    Common Methods:

    • as_uri(): Returns the path as an s3:// URI.
    • from_uri(uri): Creates a PureS3Path from an S3 URI.
    • from_bucket_key(bucket, key): Creates a PureS3Path from a bucket and key pair.
    • .bucket: Property returning the bucket name string.
    • .key: Property returning the key name string.
    from s3path import PureS3Path

# Creation
p = PureS3Path('/pypi-proxy/boto3/index.html')

# Properties
print(p.bucket) # 'pypi-proxy'
print(p.key)    # 'boto3/index.html'

# URI conversion
print(p.as_uri()) # 's3://pypi-proxy/boto3/index.html'

# From URI
p2 = PureS3Path.from_uri('s3://pypi-proxy/boto3/index.html')

# From Bucket/Key
p3 = PureS3Path.from_bucket_key('pypi-proxy', 'boto3/index.html')

  2. Use PureVersionedS3Path for versioned buckets

    master

    PureVersionedS3Path is a subclass of PureS3Path used when S3 versioning is enabled on a bucket. It includes a version_id attribute.

    Key Methods:

    • from_uri(uri, *, version_id): Creates a versioned path from an S3 URI.
    • from_bucket_key(bucket, key, *, version_id): Creates a versioned path from a bucket and key pair.

    Division Operator Behavior: When using the / operator, the resulting type depends on the operands:

    • PureVersionedS3Path / PureVersionedS3Path $\rightarrow$ PureVersionedS3Path (inherits version_id from the right operand)
    • PureS3Path / PureVersionedS3Path $\rightarrow$ PureVersionedS3Path
    • str / PureVersionedS3Path $\rightarrow$ PureVersionedS3Path
    • PureVersionedS3Path / PureS3Path $\rightarrow$ PureS3Path
    • PureVersionedS3Path / str $\rightarrow$ PureS3Path
    from s3path import PureVersionedS3Path, PureS3Path

# Creation
pv = PureVersionedS3Path('/my-bucket/file.txt', version_id='v123')

# Division behavior
ps = PureS3Path('/my-bucket/other')
result = ps / pv  # Returns a PureVersionedS3Path

  3. Configure AWS credentials via boto3.setup_default_session

    master

    Since s3path uses boto3 as its driver, it relies on standard boto3 configuration methods. s3path does not provide its own configuration API for credentials; instead, you must use boto3.setup_default_session() to specify credentials and region in your code.

    Supported boto3 configuration methods include:

    • Environment variables
    • Shared credential file (~/.aws/credentials)
    • AWS config file (~/.aws/config)
    • Assume Role provider
    • Instance metadata service (on EC2 with IAM roles)
    import boto3
from s3path import S3Path

boto3.setup_default_session(
    region_name='us-east-1',
    aws_access_key_id='<access-key>',
    aws_secret_access_key='<access-secret'>
)

bucket_path = S3Path('/pypi-proxy/')

  5. Upload a file to an S3 bucket

    master

    There are two ways to upload data to S3 using S3Path:

    1. Small files: Use .write_text() to write content directly to a remote path.
    2. Large files (Buffered): To avoid loading entire files into memory, use the .open() method with a context manager and shutil.copyfileobj to stream the data from a local file to the S3 path.
    # Method 1: Small files
from pathlib import Path
from s3path import S3Path
local_path = Path('/tmp/hello.txt')
S3Path('/my-bucket/hello.txt').write_text(local_path.read_text())

# Method 2: Large files (Buffered)
import shutil
from pathlib import Path
from s3path import S3Path
local_path = Path('/tmp/hello.txt')
remote_path = S3Path('/my-bucket/hello.txt')
with local_path.open('rb') as src, remote_path.open('wb') as dst:
    shutil.copyfileobj(src, dst)

  6. Use s3path with S3-Compatible Storage (LocalStack, MinIO)

    master

    You can use s3path with S3-compatible services like LocalStack or MinIO by creating a custom boto3 resource with a specific endpoint_url and registering that resource to a specific path using register_configuration_parameter.

    import boto3
from botocore.client import Config
from s3path import PureS3Path, register_configuration_parameter

# Define paths for configuration
default_aws_s3_path = PureS3Path('/')
local_stack_bucket_path = PureS3Path('/LocalStackBucket/')
minio_bucket_path = PureS3Path('/MinIOBucket/')

# Define boto3 s3 resources with custom endpoints
local_stack_resource = boto3.resource('s3', endpoint_url='http://localhost:4566')
minio_resource = boto3.resource(
    's3',
    endpoint_url='http://localhost:9000',
    aws_access_key_id='minio',
    aws_secret_access_key='minio123',
    config=Config(signature_version='s3v4'),
    region_name='us-east-1'
)

# Map resources to specific paths
register_configuration_parameter(default_aws_s3_path, parameters={'ServerSideEncryption': 'AES256'})
register_configuration_parameter(local_stack_bucket_path, resource=local_stack_resource)
register_configuration_parameter(minio_bucket_path, resource=minio_resource)

  7. Generate a pre-signed URL with get_presigned_url()

    master

    The get_presigned_url(expire_in=3600) method returns a URL that allows anyone to perform a GET request to retrieve the file. expire_in accepts an integer (seconds) or a timedelta object.

    Important Setup: Generating pre-signed URLs requires knowing the exact AWS region and using s3v4 as the signature version. You must register your boto3 resource with register_configuration_parameter first.

    import boto3
from botocore.config import Config
from s3path import S3Path, register_configuration_parameter

# 1. Setup the resource with s3v4 and the correct region
resource = boto3.resource(
    "s3",
    config=Config(signature_version="s3v4"),
    region_name="the aws region name"
)

# 2. Register the resource
register_configuration_parameter(S3Path("/"), resource=resource)

# 3. Use the pre-signed URL
file = S3Path("/my-bucket/toto.txt")
file.write_text("hello world")
presigned_url = file.get_presigned_url()

import requests
print(requests.get(presigned_url).content)
# b"hello world"

  8. Use S3Path for basic S3 operations

    master

    S3Path provides a pathlib-like interface for AWS S3, using boto3 as the driver. You can navigate S3

    from s3path import S3Path

# Initialize a path to a bucket or prefix
bucket_path = S3Path('/pypi-proxy/')

# List 'subdirectories' (prefixes)
subdirs = [path for path in bucket_path.iterdir() if path.is_dir()]

# Glob files (e.g., find all HTML files recursively)
html_files = list(bucket_path.glob('**/*.html'))

# Navigate using the / operator
file_path = bucket_path / 'boto3' / 'boto3-1.4.1.tar.gz'

# Query properties
exists = file_path.exists()
is_file = file_path.is_file()

# Read file content
content = file_path.read_text()

# Open file for context management
with file_path.open() as f:
    data = f.read()

  10. Download a file from S3

    master

    To download a file from S3 to your local filesystem, use the .read_text() method on the S3Path object and write the result to a local file using standard pathlib.Path.write_text().

    from pathlib import Path
from s3path import S3Path
local_path = Path('./my_local_image.jpg')
local_path.write_text(S3Path('/my-bucket/my_image_in_s3.jpg').read_text())

  12. Retrieve subfolder names in an S3 bucket

    master

    To find subfolders (prefixes) within an S3 bucket, iterate through the directory using .iterdir() on the parent path and filter for items where .is_dir() is true.

    from s3path import S3Path
for path in S3Path('/my-bucket/prefix-name-with-slash/').iterdir():
    if path.is_dir():
        print('sub folder : ', path)