Work with Amazon S3 versioned objects using an Amazon SDK - Amazon Simple Storage Service
Services or capabilities described in Amazon Web Services documentation might vary by Region. To see the differences applicable to the China Regions, see Getting Started with Amazon Web Services in China (PDF).

Work with Amazon S3 versioned objects using an Amazon SDK

The following code example shows how to:

  • Create a versioned S3 bucket.

  • Get all versions of an object.

  • Roll an object back to a previous version.

  • Delete and restore a versioned object.

  • Permanently delete all versions of an object.

SDK for Python (Boto3)

There's more on GitHub. Find the complete example and learn how to set up and run in the Amazon Code Examples Repository.

Create functions that wrap S3 actions.

def create_versioned_bucket(bucket_name, prefix): """ Creates an Amazon S3 bucket, enables it for versioning, and configures a lifecycle that expires noncurrent object versions after 7 days. Adding a lifecycle configuration to a versioned bucket is a best practice. It helps prevent objects in the bucket from accumulating a large number of noncurrent versions, which can slow down request performance. Usage is shown in the usage_demo_single_object function at the end of this module. :param bucket_name: The name of the bucket to create. :param prefix: Identifies which objects are automatically expired under the configured lifecycle rules. :return: The newly created bucket. """ try: bucket = s3.create_bucket( Bucket=bucket_name, CreateBucketConfiguration={ "LocationConstraint": s3.meta.client.meta.region_name }, )"Created bucket %s.", except ClientError as error: if error.response["Error"]["Code"] == "BucketAlreadyOwnedByYou": logger.warning("Bucket %s already exists! Using it.", bucket_name) bucket = s3.Bucket(bucket_name) else: logger.exception("Couldn't create bucket %s.", bucket_name) raise try: bucket.Versioning().enable()"Enabled versioning on bucket %s.", except ClientError: logger.exception("Couldn't enable versioning on bucket %s.", raise try: expiration = 7 bucket.LifecycleConfiguration().put( LifecycleConfiguration={ "Rules": [ { "Status": "Enabled", "Prefix": prefix, "NoncurrentVersionExpiration": {"NoncurrentDays": expiration}, } ] } ) "Configured lifecycle to expire noncurrent versions after %s days " "on bucket %s.", expiration,, ) except ClientError as error: logger.warning( "Couldn't configure lifecycle on bucket %s because %s. " "Continuing anyway.",, error, ) return bucket def rollback_object(bucket, object_key, version_id): """ Rolls back an object to an earlier version by deleting all versions that occurred after the specified rollback version. Usage is shown in the usage_demo_single_object function at the end of this module. :param bucket: The bucket that holds the object to roll back. :param object_key: The object to roll back. :param version_id: The version ID to roll back to. """ # Versions must be sorted by last_modified date because delete markers are # at the end of the list even when they are interspersed in time. versions = sorted( bucket.object_versions.filter(Prefix=object_key), key=attrgetter("last_modified"), reverse=True, ) logger.debug( "Got versions:\n%s", "\n".join( [ f"\t{version.version_id}, last modified {version.last_modified}" for version in versions ] ), ) if version_id in [ver.version_id for ver in versions]: print(f"Rolling back to version {version_id}") for version in versions: if version.version_id != version_id: version.delete() print(f"Deleted version {version.version_id}") else: break print(f"Active version is now {bucket.Object(object_key).version_id}") else: raise KeyError( f"{version_id} was not found in the list of versions for " f"{object_key}." ) def revive_object(bucket, object_key): """ Revives a versioned object that was deleted by removing the object's active delete marker. A versioned object presents as deleted when its latest version is a delete marker. By removing the delete marker, we make the previous version the latest version and the object then presents as *not* deleted. Usage is shown in the usage_demo_single_object function at the end of this module. :param bucket: The bucket that contains the object. :param object_key: The object to revive. """ # Get the latest version for the object. response = s3.meta.client.list_object_versions(, Prefix=object_key, MaxKeys=1 ) if "DeleteMarkers" in response: latest_version = response["DeleteMarkers"][0] if latest_version["IsLatest"]: "Object %s was indeed deleted on %s. Let's revive it.", object_key, latest_version["LastModified"], ) obj = bucket.Object(object_key) obj.Version(latest_version["VersionId"]).delete() "Revived %s, active version is now %s with body '%s'", object_key, obj.version_id, obj.get()["Body"].read(), ) else: logger.warning( "Delete marker is not the latest version for %s!", object_key ) elif "Versions" in response: logger.warning("Got an active version for %s, nothing to do.", object_key) else: logger.error("Couldn't get any version info for %s.", object_key) def permanently_delete_object(bucket, object_key): """ Permanently deletes a versioned object by deleting all of its versions. Usage is shown in the usage_demo_single_object function at the end of this module. :param bucket: The bucket that contains the object. :param object_key: The object to delete. """ try: bucket.object_versions.filter(Prefix=object_key).delete()"Permanently deleted all versions of object %s.", object_key) except ClientError: logger.exception("Couldn't delete all versions of %s.", object_key) raise

Upload the stanza of a poem to a versioned object and perform a series of actions on it.

def usage_demo_single_object(obj_prefix="demo-versioning/"): """ Demonstrates usage of versioned object functions. This demo uploads a stanza of a poem and performs a series of revisions, deletions, and revivals on it. :param obj_prefix: The prefix to assign to objects created by this demo. """ with open("father_william.txt") as file: stanzas ="\n\n") width = get_terminal_size((80, 20))[0] print("-" * width) print("Welcome to the usage demonstration of Amazon S3 versioning.") print( "This demonstration uploads a single stanza of a poem to an Amazon " "S3 bucket and then applies various revisions to it." ) print("-" * width) print("Creating a version-enabled bucket for the demo...") bucket = create_versioned_bucket("bucket-" + str(uuid.uuid1()), obj_prefix) print("\nThe initial version of our stanza:") print(stanzas[0]) # Add the first stanza and revise it a few times. print("\nApplying some revisions to the stanza...") obj_stanza_1 = bucket.Object(f"{obj_prefix}stanza-1") obj_stanza_1.put(Body=bytes(stanzas[0], "utf-8")) obj_stanza_1.put(Body=bytes(stanzas[0].upper(), "utf-8")) obj_stanza_1.put(Body=bytes(stanzas[0].lower(), "utf-8")) obj_stanza_1.put(Body=bytes(stanzas[0][::-1], "utf-8")) print( "The latest version of the stanza is now:", obj_stanza_1.get()["Body"].read().decode("utf-8"), sep="\n", ) # Versions are returned in order, most recent first. obj_stanza_1_versions = bucket.object_versions.filter(Prefix=obj_stanza_1.key) print( "The version data of the stanza revisions:", *[ f" {version.version_id}, last modified {version.last_modified}" for version in obj_stanza_1_versions ], sep="\n", ) # Rollback two versions. print("\nRolling back two versions...") rollback_object(bucket, obj_stanza_1.key, list(obj_stanza_1_versions)[2].version_id) print( "The latest version of the stanza:", obj_stanza_1.get()["Body"].read().decode("utf-8"), sep="\n", ) # Delete the stanza print("\nDeleting the stanza...") obj_stanza_1.delete() try: obj_stanza_1.get() except ClientError as error: if error.response["Error"]["Code"] == "NoSuchKey": print("The stanza is now deleted (as expected).") else: raise # Revive the stanza print("\nRestoring the stanza...") revive_object(bucket, obj_stanza_1.key) print( "The stanza is restored! The latest version is again:", obj_stanza_1.get()["Body"].read().decode("utf-8"), sep="\n", ) # Permanently delete all versions of the object. This cannot be undone! print("\nPermanently deleting all versions of the stanza...") permanently_delete_object(bucket, obj_stanza_1.key) obj_stanza_1_versions = bucket.object_versions.filter(Prefix=obj_stanza_1.key) if len(list(obj_stanza_1_versions)) == 0: print("The stanza has been permanently deleted and now has no versions.") else: print("Something went wrong. The stanza still exists!") print(f"\nRemoving {}...") bucket.delete() print(f"{} deleted.") print("Demo done!")

For a complete list of Amazon SDK developer guides and code examples, see Using this service with an Amazon SDK. This topic also includes information about getting started and details about previous SDK versions.