types-s3transfer

import boto3 from s3transfer.manager import S3Transfer, TransferConfig from s3transfer.futures import TransferFuture import os def upload_file_to_s3(local_filepath: str, bucket_name: str, s3_key: str) -> TransferFuture: # Initialize S3 client s3_client = boto3.client( 's3', region_name=os.environ.get('AWS_REGION', 'us-east-1'), aws_access_key_id=os.environ.get('AWS_ACCESS_KEY_ID', 'DUMMY_KEY'), aws_secret_access_key=os.environ.get('AWS_SECRET_ACCESS_KEY', 'DUMMY_SECRET') ) # Configure transfer (optional) transfer_config = TransferConfig( multipart_threshold=8 * 1024 * 1024, # 8 MB max_concurrency=10 ) # Create S3Transfer manager transfer_manager = S3Transfer(s3_client, transfer_config) # Perform upload future: TransferFuture = transfer_manager.upload_file(local_filepath, bucket_name, s3_key) print(f"Uploading {local_filepath} to s3://{bucket_name}/{s3_key}") return future if __name__ == '__main__': # Example usage (replace with actual values for testing) dummy_filepath = 'temp_file.txt' dummy_bucket = 'your-test-bucket' dummy_s3_key = 'test-upload/my_document.txt' # Create a dummy file with open(dummy_filepath, 'w') as f: f.write('Hello from types-s3transfer quickstart!') print(f"Created dummy file: {dummy_filepath}") try: # The 'types-s3transfer' package ensures that 'future' is correctly typed as TransferFuture future = upload_file_to_s3(dummy_filepath, dummy_bucket, dummy_s3_key) future.result() # Wait for the upload to complete print(f"Upload successful: s3://{dummy_bucket}/{dummy_s3_key}") except Exception as e: print(f"Upload failed: {e}") finally: # Clean up the dummy file if os.path.exists(dummy_filepath): os.remove(dummy_filepath) print(f"Removed dummy file: {dummy_filepath}") # To benefit from type hints, ensure 'types-s3transfer' is installed in your environment. # Type checkers like MyPy or Pyright will then provide suggestions and error checking.