Cleanup and adding test file.

Author: rlhagerm

python/example_code/s3/scenarios/conditional_requests/s3_conditional_requests.py

@@ -1,6 +1,7 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0
 
+import boto3
 import logging
 
 from botocore.exceptions import ClientError
@@ -12,35 +13,50 @@
 class S3ConditionalRequests:
     """Encapsulates S3 conditional request operations."""
 
-    def __init__(self, s3_client, source_bucket: str, dest_bucket: str):
+    def __init__(self, s3_client):
         self.s3 = s3_client
-        self.source_bucket = source_bucket
-        self.dest_bucket = dest_bucket
+
+    @classmethod
+    def from_client(cls):
+        """
+        Instantiates this class from a Boto3 client.
+        """
+        s3_client = boto3.client("s3")
+        return cls(s3_client)
 
     # snippet-start:[python.example_code.s3.GetObjectConditional]
 
-    def get_object_conditional(self, object_key: str, condition_type: str, condition_value: str):
+    def get_object_conditional(
+        self,
+        object_key: str,
+        source_bucket: str,
+        condition_type: str,
+        condition_value: str,
+    ):
         """
         Retrieves an object from Amazon S3 with a conditional request.
 
         :param object_key: The key of the object to retrieve.
+        :param source_bucket: The source bucket of the object.
         :param condition_type: The type of condition: 'IfMatch', 'IfNoneMatch', 'IfModifiedSince', 'IfUnmodifiedSince'.
         :param condition_value: The value to use for the condition.
         """
         try:
             response = self.s3.get_object(
-                Bucket=self.source_bucket,
+                Bucket=source_bucket,
                 Key=object_key,
-                **{condition_type: condition_value}
+                **{condition_type: condition_value},
+            )
+            sample_bytes = response["Body"].read(20)
+            print(
+                f"\tConditional read successful. Here are the first 20 bytes of the object:\n"
             )
-            sample_bytes = response['Body'].read(20)
-            print(f"\tConditional read successful. Here are the first 20 bytes of the object:\n")
             print(f"\t{sample_bytes}")
         except ClientError as e:
-            error_code = e.response['Error']['Code']
-            if error_code == '412':
+            error_code = e.response["Error"]["Code"]
+            if error_code == "PreconditionFailed":
                 print("\tConditional read failed: Precondition failed")
-            if error_code == '304':
+            elif error_code == "304":  # Not modified error code.
                 print("\tConditional read failed: Object not modified")
             else:
                 logger.error(f"Unexpected error: {error_code}")
@@ -50,26 +66,25 @@ def get_object_conditional(self, object_key: str, condition_type: str, condition
 
     # snippet-start:[python.example_code.s3.PutObjectConditional]
 
-    def put_object_conditional(self, object_key: str, data: bytes, condition_type: str, condition_value: str):
+    def put_object_conditional(self, object_key: str, source_bucket: str, data: bytes):
         """
-        Uploads an object to Amazon S3 with a conditional request.
+        Uploads an object to Amazon S3 with a conditional request. Prevents overwrite
+        using an IfNoneMatch condition for the object key.
 
         :param object_key: The key of the object to upload.
+        :param source_bucket: The source bucket of the object.
         :param data: The data to upload.
-        :param condition_type: The type of condition to apply, e.g. 'IfNoneMatch'.
-        :param condition_value: The value to use for the condition.
         """
         try:
-            response = self.s3.put_object(
-                Bucket=self.source_bucket,
-                Key=object_key,
-                Body=data,
-                **{condition_type: condition_value}
+            self.s3.put_object(
+                Bucket=source_bucket, Key=object_key, Body=data, IfNoneMatch="*"
+            )
+            print(
+                f"\tConditional write successful for key {object_key} in bucket {source_bucket}."
             )
-            return response
         except ClientError as e:
-            error_code = e.response['Error']['Code']
-            if error_code == '412':
+            error_code = e.response["Error"]["Code"]
+            if error_code == "PreconditionFailed":
                 print("\tConditional copy failed: Precondition failed")
             else:
                 logger.error(f"Unexpected error: {error_code}")
@@ -78,37 +93,44 @@ def put_object_conditional(self, object_key: str, data: bytes, condition_type: s
     # snippet-end:[python.example_code.s3.PutObjectConditional]
 
     # snippet-start:[python.example_code.s3.CopyObjectConditional]
-    def copy_object_conditional(self, source_key: str, dest_key: str, condition_type: str, condition_value: str):
+    def copy_object_conditional(
+        self,
+        source_key: str,
+        dest_key: str,
+        source_bucket: str,
+        dest_bucket: str,
+        condition_type: str,
+        condition_value: str,
+    ):
         """
         Copies an object from one Amazon S3 bucket to another with a conditional request.
 
         :param source_key: The key of the source object to copy.
         :param dest_key: The key of the destination object.
+        :param source_bucket: The source bucket of the object.
+        :param dest_bucket: The destination bucket of the object.
         :param condition_type: The type of condition to apply, e.g.
         'CopySourceIfMatch', 'CopySourceIfNoneMatch', 'CopySourceIfModifiedSince', 'CopySourceIfUnmodifiedSince'.
         :param condition_value: The value to use for the condition.
         """
         try:
-            response = self.s3.copy_object(
-                Bucket=self.dest_bucket,
+            self.s3.copy_object(
+                Bucket=dest_bucket,
                 Key=dest_key,
-                CopySource={'Bucket': self.source_bucket, 'Key': source_key},
-                **{condition_type: condition_value}
+                CopySource={"Bucket": source_bucket, "Key": source_key},
+                **{condition_type: condition_value},
+            )
+            print(
+                f"\tConditional copy successful for key {dest_key} in bucket {dest_bucket}."
             )
-            sample_bytes = response['Body'].read(20)
-            print(f"\tConditional copy successful. Here are the first 20 bytes of the object:\n")
-            print(f"\t{sample_bytes}")
         except ClientError as e:
-            error_code = e.response['Error']['Code']
-            if error_code == '412':
+            error_code = e.response["Error"]["Code"]
+            if error_code == "PreconditionFailed":
                 print("\tConditional copy failed: Precondition failed")
-            if error_code == '304':
+            elif error_code == "304":  # Not modified error code.
                 print("\tConditional copy failed: Object not modified")
             else:
                 logger.error(f"Unexpected error: {error_code}")
                 raise
 
     # snippet-end:[python.example_code.s3.CopyObjectConditional]
-
-
-