Skip to content
Search

Boto3

lastmod 2024-05-06created 2023-02-228 edits

AWS SDK for Python.

Get started quickly with AWS using boto3, the AWS SDK for Python. Boto3 makes it easy to integrate your Python application, library, or script with AWS services including Amazon S3, Amazon EC2, Amazon DynamoDB, and more.

Client APIs

  • S3 - Boto3 1.34.98 documentation
  • abort_multipart_upload
  • can_paginate
  • close
  • complete_multipart_upload
  • copy
  • copy_object
  • create_bucket
  • create_multipart_upload
  • create_session
  • delete_bucket
  • delete_bucket_analytics_configuration
  • delete_bucket_cors
  • delete_bucket_encryption
  • delete_bucket_intelligent_tiering_configuration
  • delete_bucket_inventory_configuration
  • delete_bucket_lifecycle
  • delete_bucket_metrics_configuration
  • delete_bucket_ownership_controls
  • delete_bucket_policy
  • delete_bucket_replication
  • delete_bucket_tagging
  • delete_bucket_website
  • delete_object
  • delete_object_tagging
  • delete_objects
  • delete_public_access_block
  • download_file
  • download_fileobj
  • generate_presigned_post
  • generate_presigned_url
  • get_bucket_accelerate_configuration
  • get_bucket_acl
  • get_bucket_analytics_configuration
  • get_bucket_cors
  • get_bucket_encryption
  • get_bucket_intelligent_tiering_configuration
  • get_bucket_inventory_configuration
  • get_bucket_lifecycle
  • get_bucket_lifecycle_configuration
  • get_bucket_location
  • get_bucket_logging
  • get_bucket_metrics_configuration
  • get_bucket_notification
  • get_bucket_notification_configuration
  • get_bucket_ownership_controls
  • get_bucket_policy
  • get_bucket_policy_status
  • get_bucket_replication
  • get_bucket_request_payment
  • get_bucket_tagging
  • get_bucket_versioning
  • get_bucket_website
  • get_object
  • get_object_acl
  • get_object_attributes
  • get_object_legal_hold
  • get_object_lock_configuration
  • get_object_retention
  • get_object_tagging
  • get_object_torrent
  • get_paginator
  • get_public_access_block
  • get_waiter
  • head_bucket
  • head_object
  • list_bucket_analytics_configurations
  • list_bucket_intelligent_tiering_configurations
  • list_bucket_inventory_configurations
  • list_bucket_metrics_configurations
  • list_buckets
  • list_directory_buckets
  • list_multipart_uploads
  • list_object_versions
  • list_objects
  • list_objects_v2
  • list_parts
  • put_bucket_accelerate_configuration
  • put_bucket_acl
  • put_bucket_analytics_configuration
  • put_bucket_cors
  • put_bucket_encryption
  • put_bucket_intelligent_tiering_configuration
  • put_bucket_inventory_configuration
  • put_bucket_lifecycle
  • put_bucket_lifecycle_configuration
  • put_bucket_logging
  • put_bucket_metrics_configuration
  • put_bucket_notification
  • put_bucket_notification_configuration
  • put_bucket_ownership_controls
  • put_bucket_policy
  • put_bucket_replication
  • put_bucket_request_payment
  • put_bucket_tagging
  • put_bucket_versioning
  • put_bucket_website
  • put_object
  • put_object_acl
  • put_object_legal_hold
  • put_object_lock_configuration
  • put_object_retention
  • put_object_tagging
  • put_public_access_block
  • restore_object
  • select_object_content
  • upload_file
  • upload_fileobj
  • upload_part
  • upload_part_copy
  • write_get_object_response

Key Exists 

import boto3
import botocore

s3 = boto3.resource('s3')

try:
    s3.Object('my-bucket', 'dootdoot.jpg').load()
except botocore.exceptions.ClientError as e:
    if e.response['Error']['Code'] == "404":
        # The object does not exist.
        ...
    else:
        # Something else has gone wrong.
        raise
else:
    # The object does exist.
    ...

How to run boto3 with asyncio

Must be run with Threadpool-based MultiThread. 

from pprint import pprint
from concurrent.futures import ThreadPoolExecutor
import asyncio
import boto3

_executor = ThreadPoolExecutor(10)

async def execute(region):
    print(f"call the region {region}")

    loop = asyncio.get_running_loop()
    ec2 = boto3.client("ec2", region_name=region)
    response = await loop.run_in_executor(_executor, ec2.describe_instances)
    return response

async def main():
    ec2 = boto3.client("ec2", region_name="ap-northeast-2")
    regions = [region['RegionName'] for region in ec2.describe_regions()['Regions']]

    task_list = [asyncio.ensure_future(execute(region)) for region in regions]
    done, pending = await asyncio.wait(task_list)

    results = [d.result() for d in done]
    return results

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(main())
    pprint(result)

Cloudflare R2 Example 

import boto3

s3 = boto3.client(
    service_name ="s3",
    endpoint_url = 'https://<accountid>.r2.cloudflarestorage.com',
    aws_access_key_id = '<access_key_id>',
    aws_secret_access_key = '<access_key_secret>'
    region_name="<location>", # Must be one of: wnam, enam, weur, eeur, apac, auto
)

# Get object information
object_information = s3.head_object(Bucket=<R2_BUCKET_NAME>, Key=<FILE_KEY_NAME>)

# Upload/Update single file
s3.upload_fileobj(io.BytesIO(file_content), <R2_BUCKET_NAME>, <FILE_KEY_NAME>)

# Delete object
s3.delete_object(Bucket=<R2_BUCKET_NAME>, Key=<FILE_KEY_NAME>)

See also

Favorite site

author Zer0 sitemap