Amazon S3 Control examples using SDK for Python (Boto3)

The following code examples show you how to perform actions and implement common scenarios by using the AWS SDK for Python (Boto3) with Amazon S3 Control.

Basics are code examples that show you how to perform the essential operations within a service.

Actions are code excerpts from larger programs and must be run in context. While actions show you how to call individual service functions, you can see actions in context in their related scenarios.

Each example includes a link to the complete source code, where you can find instructions on how to set up and run the code in context.

Get started

Hello Amazon S3 Control

The following code examples show how to get started using Amazon S3 Control.

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.

    def list_jobs(self, account_id: str) -> None:
        """
        List all batch jobs for the account.

        Args:
            account_id (str): AWS account ID
        """
        try:
            response = self.s3control_client.list_jobs(
                AccountId=account_id,
                JobStatuses=['Active', 'Complete', 'Cancelled', 'Failed', 'New', 'Paused', 'Pausing', 'Preparing', 'Ready', 'Suspended']
            )
            jobs = response.get('Jobs', [])
            for job in jobs:
                print(f"The job id is {job['JobId']}")
                print(f"The job priority is {job['Priority']}")
        except ClientError as e:
            print(f"Error listing jobs: {e}")
            raise

• For API details, see ListJobs in AWS SDK for Python (Boto3) API Reference.

Basics

Learn the basics

The following code example shows how to learn core operations for Amazon S3 Control.

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.

Learn S3 Batch Basics Scenario.

class S3BatchScenario:
    """Manages the S3 Batch Operations scenario."""

    DASHES = "-" * 80
    STACK_NAME = "MyS3Stack"

    def __init__(self, s3_batch_wrapper: S3BatchWrapper, cfn_helper: CloudFormationHelper) -> None:
        """
        Initialize the S3 Batch scenario.

        Args:
            s3_batch_wrapper: S3BatchWrapper instance
            cfn_helper: CloudFormationHelper instance
        """
        self.s3_batch_wrapper = s3_batch_wrapper
        self.cfn_helper = cfn_helper

    def wait_for_input(self) -> None:
        """Wait for user input to continue."""
        q.ask("\nPress Enter to continue...")
        print()

    def setup_resources(self, bucket_name: str, file_names: list) -> Tuple[str, str]:
        """
        Set up initial resources for the scenario.

        Args:
            bucket_name (str): Name of the bucket to create
            file_names (list): List of files to upload

        Returns:
            tuple: Manifest location and report bucket ARN
        """
        print("\nSetting up required resources...")
        self.s3_batch_wrapper.create_bucket(bucket_name)
        report_bucket_arn = f"arn:aws:s3:::{bucket_name}"
        manifest_location = f"arn:aws:s3:::{bucket_name}/job-manifest.csv"
        self.s3_batch_wrapper.upload_files_to_bucket(bucket_name, file_names)
        return manifest_location, report_bucket_arn

    def run_scenario(self) -> None:
        """Run the S3 Batch Operations scenario."""
        account_id = self.s3_batch_wrapper.get_account_id()
        bucket_name = f"demo-s3-batch-{str(uuid.uuid4())}"
        file_names = [
            "job-manifest.csv",
            "object-key-1.txt",
            "object-key-2.txt",
            "object-key-3.txt",
            "object-key-4.txt"
        ]

        print(self.DASHES)
        print("Welcome to the Amazon S3 Batch basics scenario.")
        print("""
    S3 Batch operations enables efficient and cost-effective processing of large-scale 
    data stored in Amazon S3. It automatically scales resources to handle varying workloads 
    without the need for manual intervention.
    
    This Python program walks you through Amazon S3 Batch operations.
    """)

        try:
            # Deploy CloudFormation stack for IAM roles
            print("Deploying CloudFormation stack...")
            self.cfn_helper.deploy_cloudformation_stack(self.STACK_NAME)
            stack_outputs = self.cfn_helper.get_stack_outputs(self.STACK_NAME)
            iam_role_arn = stack_outputs.get('S3BatchRoleArn')
            
            # Set up S3 bucket and upload test files
            manifest_location, report_bucket_arn = self.setup_resources(
                bucket_name, file_names
            )

            self.wait_for_input()

            print("\n1. Creating S3 Batch Job...")
            job_id = self.s3_batch_wrapper.create_s3_batch_job(
                account_id,
                iam_role_arn,
                manifest_location,
                report_bucket_arn
            )

            time.sleep(5)
            failure_reasons = self.s3_batch_wrapper.check_job_failure_reasons(job_id, account_id)
            if failure_reasons:
                print("\nJob failed. Please fix the issues and try again.")
                if not q.ask(
                    "Do you want to proceed with the rest of the operations? (y/n): ", q.is_yesno
                ):
                    raise ValueError("Job failed, stopping execution")

            self.wait_for_input()
            print("\n" + self.DASHES)
            print("2. Update an existing S3 Batch Operations job's priority")
            print("In this step, we modify the job priority value. The higher the number, the higher the priority.")
            self.s3_batch_wrapper.update_job_priority(job_id, account_id)
            
            self.wait_for_input()
            print("\n" + self.DASHES)
            print("3. Cancel the S3 Batch job")
            cancel_job = q.ask("Do you want to cancel the Batch job? (y/n): ", q.is_yesno)
            if cancel_job:
                self.s3_batch_wrapper.cancel_job(job_id, account_id)
            else:
                print(f"Job {job_id} was not canceled.")
                
            self.wait_for_input()
            print("\n" + self.DASHES)
            print("4. Describe the job that was just created")
            self.s3_batch_wrapper.describe_job_details(job_id, account_id)
            
            self.wait_for_input()
            print("\n" + self.DASHES)
            print("5. Describe the tags associated with the job")
            self.s3_batch_wrapper.get_job_tags(job_id, account_id)
            
            self.wait_for_input()
            print("\n" + self.DASHES)
            print("6. Update Batch Job Tags")
            self.s3_batch_wrapper.put_job_tags(job_id, account_id)
            
            self.wait_for_input()
            print("\n" + self.DASHES)
            print("7. List Batch Jobs")
            self.s3_batch_wrapper.list_jobs(account_id)
            
            self.wait_for_input()
            print("\n" + self.DASHES)
            print("8. Delete the Amazon S3 Batch job tagging")
            delete_tags = q.ask("Do you want to delete Batch job tagging? (y/n): ", q.is_yesno)
            if delete_tags:
                self.s3_batch_wrapper.delete_job_tags(job_id, account_id)

            print("\n" + self.DASHES)
            if q.ask(
                "Do you want to delete the AWS resources used in this scenario? (y/n): ", q.is_yesno
            ):
                self.s3_batch_wrapper.cleanup_resources(bucket_name, file_names)
                self.cfn_helper.destroy_cloudformation_stack(self.STACK_NAME)

        except Exception as e:
            print(f"An error occurred: {e}")
            print("\nCleaning up resources due to failure...")
            try:
                self.s3_batch_wrapper.cleanup_resources(bucket_name, file_names)
                self.cfn_helper.destroy_cloudformation_stack(self.STACK_NAME)
            except Exception as cleanup_error:
                print(f"Error during cleanup: {cleanup_error}")
            raise

        print("\nThe Amazon S3 Batch scenario has successfully completed.")
        print(self.DASHES)

• For API details, see the following topics in AWS SDK for Python (Boto3) API Reference.

  • CreateJob

  • DeleteJobTagging

  • DescribeJob

  • GetJobTagging

  • ListJobs

  • PutJobTagging

  • UpdateJobPriority

  • UpdateJobStatus

Actions

CreateJob

The following code example shows how to use CreateJob.

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.

    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

• For API details, see CreateJob in AWS SDK for Python (Boto3) API Reference.

DeleteJobTagging

The following code example shows how to use DeleteJobTagging.

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.

    def delete_job_tags(self, job_id: str, account_id: str) -> None:
        """
        Delete all tags from a batch job.

        Args:
            job_id (str): ID of the batch job
            account_id (str): AWS account ID
        """
        try:
            self.s3control_client.delete_job_tagging(
                AccountId=account_id,
                JobId=job_id
            )
            print(f"You have successfully deleted {job_id} tagging.")
        except ClientError as e:
            print(f"Error deleting job tags: {e}")
            raise

• For API details, see DeleteJobTagging in AWS SDK for Python (Boto3) API Reference.

DescribeJob

The following code example shows how to use DescribeJob.

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.

    def describe_job_details(self, job_id: str, account_id: str) -> None:
        """
        Describe detailed information about a batch job.

        Args:
            job_id (str): ID of the batch job
            account_id (str): AWS account ID
        """
        try:
            response = self.s3control_client.describe_job(
                AccountId=account_id,
                JobId=job_id
            )
            job = response['Job']
            print(f"Job ID: {job['JobId']}")
            print(f"Description: {job.get('Description', 'N/A')}")
            print(f"Status: {job['Status']}")
            print(f"Role ARN: {job['RoleArn']}")
            print(f"Priority: {job['Priority']}")
            if 'ProgressSummary' in job:
                progress = job['ProgressSummary']
                print(f"Progress Summary: Total={progress.get('TotalNumberOfTasks', 0)}, "
                      f"Succeeded={progress.get('NumberOfTasksSucceeded', 0)}, "
                      f"Failed={progress.get('NumberOfTasksFailed', 0)}")
        except ClientError as e:
            print(f"Error describing job: {e}")
            raise

• For API details, see DescribeJob in AWS SDK for Python (Boto3) API Reference.

GetJobTagging

The following code example shows how to use GetJobTagging.

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.

    def get_job_tags(self, job_id: str, account_id: str) -> None:
        """
        Get tags associated with a batch job.

        Args:
            job_id (str): ID of the batch job
            account_id (str): AWS account ID
        """
        try:
            response = self.s3control_client.get_job_tagging(
                AccountId=account_id,
                JobId=job_id
            )
            tags = response.get('Tags', [])
            if tags:
                print(f"Tags for job {job_id}:")
                for tag in tags:
                    print(f"  {tag['Key']}: {tag['Value']}")
            else:
                print(f"No tags found for job ID: {job_id}")
        except ClientError as e:
            print(f"Error getting job tags: {e}")
            raise

• For API details, see GetJobTagging in AWS SDK for Python (Boto3) API Reference.

PutJobTagging

The following code example shows how to use PutJobTagging.

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.

    def put_job_tags(self, job_id: str, account_id: str) -> None:
        """
        Add tags to a batch job.

        Args:
            job_id (str): ID of the batch job
            account_id (str): AWS account ID
        """
        try:
            self.s3control_client.put_job_tagging(
                AccountId=account_id,
                JobId=job_id,
                Tags=[
                    {'Key': 'Environment', 'Value': 'Development'},
                    {'Key': 'Team', 'Value': 'DataProcessing'}
                ]
            )
            print(f"Additional tags were added to job {job_id}")
        except ClientError as e:
            print(f"Error adding job tags: {e}")
            raise

• For API details, see PutJobTagging in AWS SDK for Python (Boto3) API Reference.

UpdateJobPriority

The following code example shows how to use UpdateJobPriority.

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.

    def update_job_priority(self, job_id: str, account_id: str) -> None:
        """
        Update the priority of a batch job and start it.

        Args:
            job_id (str): ID of the batch job
            account_id (str): AWS account ID
        """
        try:
            response = self.s3control_client.describe_job(
                AccountId=account_id,
                JobId=job_id
            )
            current_status = response['Job']['Status']
            print(f"Current job status: {current_status}")
            
            if current_status in ['Ready', 'Suspended']:
                self.s3control_client.update_job_priority(
                    AccountId=account_id,
                    JobId=job_id,
                    Priority=60
                )
                print("The job priority was updated")
                
                try:
                    self.s3control_client.update_job_status(
                        AccountId=account_id,
                        JobId=job_id,
                        RequestedJobStatus='Ready'
                    )
                    print("Job activated successfully")
                except ClientError as activation_error:
                    print(f"Note: Could not activate job automatically: {activation_error}")
                    print("Job priority was updated successfully. Job may need manual activation in the console.")
            elif current_status in ['Active', 'Completing', 'Complete']:
                print(f"Job is in '{current_status}' state - priority cannot be updated")
                if current_status == 'Completing':
                    print("Job is finishing up and will complete soon.")
                elif current_status == 'Complete':
                    print("Job has already completed successfully.")
                else:
                    print("Job is currently running.")
            else:
                print(f"Job is in '{current_status}' state - priority update not allowed")
                
        except ClientError as e:
            print(f"Error updating job priority: {e}")
            print("Continuing with the scenario...")
            return

• For API details, see UpdateJobPriority in AWS SDK for Python (Boto3) API Reference.

UpdateJobStatus

The following code example shows how to use UpdateJobStatus.

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.

    def cancel_job(self, job_id: str, account_id: str) -> None:
        """
        Cancel an S3 batch job.

        Args:
            job_id (str): ID of the batch job
            account_id (str): AWS account ID
        """
        try:
            response = self.s3control_client.describe_job(
                AccountId=account_id,
                JobId=job_id
            )
            current_status = response['Job']['Status']
            print(f"Current job status: {current_status}")

            if current_status in ['Ready', 'Suspended', 'Active']:
                self.s3control_client.update_job_status(
                    AccountId=account_id,
                    JobId=job_id,
                    RequestedJobStatus='Cancelled'
                )
                print(f"Job {job_id} was successfully canceled.")
            elif current_status in ['Completing', 'Complete']:
                print(f"Job is in '{current_status}' state - cannot be cancelled")
                if current_status == 'Completing':
                    print("Job is finishing up and will complete soon.")
                elif current_status == 'Complete':
                    print("Job has already completed successfully.")
            else:
                print(f"Job is in '{current_status}' state - cancel not allowed")
        except ClientError as e:
            print(f"Error canceling job: {e}")
            raise

• For API details, see UpdateJobStatus in AWS SDK for Python (Boto3) API Reference.