AWS S3 Utilities

airtable_apply_annotations.s3_utils

bulk_s3_actions(action, bucket, files, sources, targets=None)

Applies a bulk S3 CRUD action for all files in sources and optionally, targets.

Parameters:

Name	Type	Description	Default
action	Callable	Function calling an S3 CRUD operation.	required
bucket	str	S3 bucket name.	required
files	List[str]	List of files in bucket/sources	required
sources	List[str]	Source folders in bucket.	required
targets	List[str]	Target folders in bucket. Defaults to None.	None

Source code in src/airtable_apply_annotations/s3_utils.py

def bulk_s3_actions(
    action: Callable,
    bucket: str,
    files: List[str],
    sources: List[str],
    targets: List[str] = None,
):
    """Applies a bulk S3 CRUD action for all `files` in `sources` and optionally, `targets`.

    Args:
        action (Callable): Function calling an S3 CRUD operation.
        bucket (str): S3 bucket name.
        files (List[str]): List of files in `bucket`/`sources`
        sources (List[str]): Source folders in `bucket`.
        targets (List[str], optional): Target folders in `bucket`. Defaults to None.
    """
    if targets:
        assert len(sources) == len(targets)
        for source, target in zip(sources, targets):
            for file in files:
                action(bucket=bucket, file=file, source=source, destination=target)
    else:
        for source in sources:
            for file in files:
                action(bucket=bucket, file=file, source=source)

copy_file(bucket, file, source, destination)

Copy file in bucket from source to destination folder.

Parameters:

Name	Type	Description	Default
bucket	str	S3 bucket name.	required
file	str	Name of file to be copied (without full-path).	required
source	str	Source folder in S3 bucket.	required
destination	str	Destination folder in S3 bucket.	required

Source code in src/airtable_apply_annotations/s3_utils.py

def copy_file(bucket: str, file: str, source: str, destination: str):
    """Copy `file` in `bucket` from `source` to `destination` folder.

    Args:
        bucket (str): S3 bucket name.
        file (str): Name of file to be copied (without full-path).
        source (str): Source folder in S3 bucket.
        destination (str): Destination folder in S3 bucket.
    """
    try:
        s3_resource.Object(bucket, f"{destination}/{file}").copy_from(
            CopySource=f"{bucket}/{source}/{file}"
        )
        print(
            f"Copied file from {bucket}/{source}/{file} to",
            f"{bucket}/{destination}/{file}",
        )
    except ClientError:
        return

delete_file(bucket, file, source)

Delete file in bucket from source folder.

Parameters:

Name	Type	Description	Default
bucket	str	S3 bucket name.	required
file	str	Name of file to be deleted (without full-path).	required
source	str	Source folder in S3 bucket.	required

Source code in src/airtable_apply_annotations/s3_utils.py

def delete_file(bucket: str, file: str, source: str):
    """Delete `file` in `bucket` from `source` folder.

    Args:
        bucket (str): S3 bucket name.
        file (str): Name of file to be deleted (without full-path).
        source (str): Source folder in S3 bucket.
    """
    try:
        s3_resource.Object(bucket, f"{source}/{file}").delete()
        # print(f"Deleted file from {bucket}/{source}/{file}")
    except ClientError as exc:
        print(f"Failed to delete {bucket}/{source}/{file}")
        print(exc)

move_file(bucket, file, source, destination)

Move file in bucket from source to destination folder

Parameters:

Name	Type	Description	Default
bucket	str	S3 bucket name.	required
file	str	Name of file to be moved (without full-path).	required
source	str	Source folder in S3 bucket.	required
destination	str	Destination folder in S3 bucket.	required

Source code in src/airtable_apply_annotations/s3_utils.py

def move_file(bucket: str, file: str, source: str, destination: str):
    """Move `file` in `bucket` from `source` to `destination` folder

    Args:
        bucket (str): S3 bucket name.
        file (str): Name of file to be moved (without full-path).
        source (str): Source folder in S3 bucket.
        destination (str): Destination folder in S3 bucket.
    """
    try:
        s3_resource.Object(bucket, f"{destination}/{file}").copy_from(
            CopySource=f"{bucket}/{source}/{file}"
        )
        s3_resource.Object(bucket, f"{source}/{file}").delete()
    except ClientError as exc:
        print(
            f"Failed to move file {bucket}/{source}/{file} to",
            f"{bucket}/{destination}/{file}",
        )
        print(exc)

write_file(bucket, file_content, destination, save_file_name)

Writes file_content to save_file_name to bucket at destination folder.

Parameters:

Name	Type	Description	Default
bucket	str	S3 bucket name.	required
file_content	str	Content of file to write.	required
destination	str	Destination folder in S3 bucket.	required
save_file_name	str	Save file name	required

Source code in src/airtable_apply_annotations/s3_utils.py

def write_file(bucket: str, file_content: str, destination: str, save_file_name: str):
    """Writes `file_content` to `save_file_name` to `bucket` at `destination` folder.

    Args:
        bucket (str): S3 bucket name.
        file_content (str): Content of file to write.
        destination (str): Destination folder in S3 bucket.
        save_file_name (str): Save file name
    """
    save_path = f"{destination}/{save_file_name}"
    try:
        s3_client.put_object(Body=file_content, Bucket=bucket, Key=save_path)
    except ClientError:
        return