AWS code-library documentation change
Summary
Added Python code example implementation for S3 Batch Operations including job creation, tagging, status monitoring, and cleanup functionality
Security assessment
The changes add general SDK usage examples for S3 Batch Operations without mentioning specific security vulnerabilities or security controls. While the code demonstrates IAM role usage (security best practice), this is standard AWS SDK implementation guidance rather than addressing a specific security issue.
Diff
diff --git a/code-library/latest/ug/s3-control_example_s3-control_Basics_section.md b/code-library/latest/ug/s3-control_example_s3-control_Basics_section.md index 5cf85b831..7388017c0 100644 --- a//code-library/latest/ug/s3-control_example_s3-control_Basics_section.md +++ b//code-library/latest/ug/s3-control_example_s3-control_Basics_section.md @@ -9 +9 @@ There are more AWS SDK examples available in the [AWS Doc SDK Examples](https:// -The following code example shows how to learn core operations for Amazon S3 Control. +The following code examples show how to learn core operations for Amazon S3 Control. @@ -937,0 +938,507 @@ An action class that wraps operations. +Python + + +**SDK for Python (Boto3)** + + +###### Note + +There's more on GitHub. Find the complete example and learn how to set up and run in the [AWS Code Examples Repository](https://github.com/awsdocs/aws-doc-sdk-examples/tree/main/python/example_code/s3/scenarios/batch#code-examples). + +Learn S3 Batch Basics Scenario. + + + class S3BatchWrapper: + """Wrapper class for managing S3 Batch Operations.""" + + def __init__(self, s3_client: Any, s3control_client: Any, sts_client: Any) -> None: + """ + Initializes the S3BatchWrapper with AWS service clients. + + :param s3_client: A Boto3 Amazon S3 client. This client provides low-level + access to AWS S3 services. + :param s3control_client: A Boto3 Amazon S3 Control client. This client provides + low-level access to AWS S3 Control services. + :param sts_client: A Boto3 AWS STS client. This client provides low-level + access to AWS STS services. + """ + self.s3_client = s3_client + self.s3control_client = s3control_client + self.sts_client = sts_client + # Get region from the client for bucket creation logic + self.region_name = self.s3_client.meta.region_name + + def get_account_id(self) -> str: + """ + Get AWS account ID. + + Returns: + str: AWS account ID + """ + return self.sts_client.get_caller_identity()["Account"] + + def create_bucket(self, bucket_name: str) -> None: + """ + Create an S3 bucket. + + Args: + bucket_name (str): Name of the bucket to create + + Raises: + ClientError: If bucket creation fails + """ + try: + if self.region_name and self.region_name != 'us-east-1': + self.s3_client.create_bucket( + Bucket=bucket_name, + CreateBucketConfiguration={ + 'LocationConstraint': self.region_name + } + ) + else: + self.s3_client.create_bucket(Bucket=bucket_name) + print(f"Created bucket: {bucket_name}") + except ClientError as e: + print(f"Error creating bucket: {e}") + raise + + def upload_files_to_bucket(self, bucket_name: str, file_names: List[str]) -> str: + """ + Upload files to S3 bucket including manifest file. + + Args: + bucket_name (str): Target bucket name + file_names (list): List of file names to upload + + Returns: + str: ETag of the manifest file + + Raises: + ClientError: If file upload fails + """ + try: + for file_name in file_names: + if file_name != "job-manifest.csv": + content = f"Content for {file_name}" + self.s3_client.put_object( + Bucket=bucket_name, + Key=file_name, + Body=content.encode('utf-8') + ) + print(f"Uploaded {file_name} to {bucket_name}") + + manifest_content = "" + for file_name in file_names: + if file_name != "job-manifest.csv": + manifest_content += f"{bucket_name},{file_name}\n" + + manifest_response = self.s3_client.put_object( + Bucket=bucket_name, + Key="job-manifest.csv", + Body=manifest_content.encode('utf-8') + ) + print(f"Uploaded manifest file to {bucket_name}") + print(f"Manifest content:\n{manifest_content}") + return manifest_response['ETag'].strip('"') + + except ClientError as e: + print(f"Error uploading files: {e}") + raise + + def create_s3_batch_job(self, account_id: str, role_arn: str, manifest_location: str, + report_bucket_name: str) -> str: + """ + Create an S3 batch operation job. + + Args: + account_id (str): AWS account ID + role_arn (str): IAM role ARN for batch operations + manifest_location (str): Location of the manifest file + report_bucket_name (str): Bucket for job reports + + Returns: + str: Job ID + + Raises: + ClientError: If job creation fails + """ + try: + bucket_name = manifest_location.split(':::')[1].split('/')[0] + manifest_key = 'job-manifest.csv' + manifest_obj = self.s3_client.head_object( + Bucket=bucket_name, + Key=manifest_key + ) + etag = manifest_obj['ETag'].strip('"') + + response = self.s3control_client.create_job( + AccountId=account_id, + Operation={ + 'S3PutObjectTagging': { + 'TagSet': [ + { + 'Key': 'BatchTag', + 'Value': 'BatchValue' + }, + ] + } + }, + Report={ + 'Bucket': report_bucket_name, + 'Format': 'Report_CSV_20180820', + 'Enabled': True, + 'Prefix': 'batch-op-reports', + 'ReportScope': 'AllTasks' + }, + Manifest={ + 'Spec': { + 'Format': 'S3BatchOperations_CSV_20180820', + 'Fields': ['Bucket', 'Key'] + }, + 'Location': { + 'ObjectArn': manifest_location, + 'ETag': etag + } + }, + Priority=10, + RoleArn=role_arn, + Description='Batch job for tagging objects', + ConfirmationRequired=True + ) + job_id = response['JobId'] + print(f"The Job id is {job_id}") + return job_id + except ClientError as e: + print(f"Error creating batch job: {e}") + if 'Message' in str(e): + print(f"Detailed error message: {e.response['Message']}") + raise + + def check_job_failure_reasons(self, job_id: str, account_id: str) -> List[Dict[str, Any]]: + """ + Check for any failure reasons of a batch job. + + Args: + job_id (str): ID of the batch job + account_id (str): AWS account ID + + Returns: + list: List of failure reasons + + Raises: + ClientError: If checking job failure reasons fails