S3 Gateway¶

Object storage operations with retry logic and type detection.

S3ServiceGateway ¶

Python

S3ServiceGateway(config: Optional[Config] = None)

Provides methods to interact with AWS S3, including uploading, downloading, moving, and deleting objects. Ensures that a single instance is used throughout the application via the Singleton pattern.

Initializes the S3ServiceGateway by setting up the S3 client using the AwsClientHub.

check_bucket_permissions ¶

Python

check_bucket_permissions(bucket_name: str) -> dict

Checks the permissions of an S3 bucket.

RETURNS	DESCRIPTION
`dict`	The access control list (ACL) of the bucket.

check_object_existence ¶

Python

check_object_existence(bucket_name: str, object_key: str) -> bool

Checks if an object exists in an S3 bucket.

RETURNS	DESCRIPTION
`bool`	True if the object exists, False otherwise.

copy_object ¶

Python

copy_object(src_bucket_name: str, src_object_key: str, dst_bucket_name: str, dst_object_key: str) -> None

Copies an object from one S3 bucket to another.

delete_object ¶

Python

delete_object(bucket_name: str, object_key: str) -> None

Deletes an object from S3.

download_object ¶

Python

download_object(bucket_name: str, object_key: str, local_path: str) -> None

Downloads an object from S3 to a local file.

get_object ¶

Python

get_object(bucket_name: str, object_key: str) -> io.BytesIO

Retrieves an object from S3 and returns its content as a file stream.

RETURNS	DESCRIPTION
`io.BytesIO`	The content of the object as a BytesIO stream.

get_object_headers ¶

Python

get_object_headers(bucket_name: str, object_key: str) -> dict

Retrieves the headers of an object in S3.

RETURNS	DESCRIPTION
`dict`	The headers of the object.

get_signed_url ¶

Python

get_signed_url(bucket_name: str, object_key: str, expiration_seconds: int = 3600) -> str

Generate a signed URL for an S3 object.

list_buckets ¶

Python

list_buckets() -> list

Lists all S3 buckets.

RETURNS	DESCRIPTION
`list`	A list of bucket names.

list_objects ¶

Python

list_objects(bucket_name: str, prefix: str = None) -> list

Lists objects in an S3 bucket, optionally filtered by a prefix.

RETURNS	DESCRIPTION
`list`	A list of object keys.

move_object ¶

Python

move_object(src_bucket_name: str, src_object_key: str, dst_bucket_name: str, dst_object_key: str) -> None

Moves an object from one S3 bucket to another.

upload_file ¶

Python

upload_file(file: Union[str, Path, bytes, BytesIO, StreamingBody], bucket_name: str, object_key: str, return_url: bool = False) -> Union[None, str]

Uploads a file to S3. Handles file paths, bytes, file-like objects, and StreamingBody.

RETURNS	DESCRIPTION
`Union[None, str]`	The S3 URL of the uploaded file if `return_url` is True, otherwise None.

verify_and_correct_extension ¶

Python

verify_and_correct_extension(object_key: str, mime_type: str) -> str

Verify the extension of the object key and correct it if necessary.