Using Python & boto3

Run the adapter

Start the adapter with the binary installed and flags described in the installation page.

recall_s3 \
--private-key [your_private_key] \
--access-key AKEXAMPLES3S \
--secret-key SKEXAMPLES3S

This will start the adapter on port 8014 by default. Also, recall the --network flag can, optionally, be provided to point to a specific Recall chain environment, and logging can be enabled by passing the -v flag.

You MUST own tokens and credits to create a bucket and store data, respectively. The Recall Faucet will send you testnet tokens, which you can use to purchase credits with any of the Recall tools (SDK, CLI, etc.). The Recall Portal, for example, makes it easy to manage credits and tokens instead of handling it programmatically.

Install boto3

Assuming you've already set up a Python environment, install boto3:

pip install boto3

Create helper functions

Create a connection

First, we'll import the necessary modules and define a helper function to connect to the adapter:

from boto3 import client as S3Client, session as S3Session
from botocore.config import Config
from botocore.exceptions import NoCredentialsError, PartialCredentialsError
 
# Establish a connection to the Recall S3 adapter
def connect_to_recall(host: str, access_key: str, secret_key: str) -> S3Client:
    try:
        # Create a session and S3 client
        session = S3Session.Session()
        recall = session.client(
            "s3",
            endpoint_url=host,
            aws_access_key_id=access_key,
            aws_secret_access_key=secret_key,
            config=Config(
                retries={"max_attempts": 3},
                s3={"addressing_style": "path"},
            ),
        )
        return recall
    except (NoCredentialsError, PartialCredentialsError, Exception) as e:
        print(f"Failed to connect to Recall: {str(e)}")
        raise

Create a bucket

Our bucket creation helper does one important thing: it prefixes the bucket name with the creator's public key. This is optional, as described in the bucket usage docs. You could choose to simply use the alias name you define upon creation.

# Create a bucket in Recall
def create_bucket(client: S3Client, bucket_name: str, owner_address: str) -> str | None:
    try:
        # Create a bucket (the name doesn't matter)
        client.create_bucket(Bucket=bucket_name)
        buckets = client.list_buckets()
        # Find the bucket with matching name
        matching_bucket = next(
            (
                bucket["Name"]
                for bucket in buckets["Buckets"]
                if bucket["Name"] == bucket_name
            ),
            None,
        )
        if matching_bucket:
            # Alter the `Name` property to prefix the name: `<owner_address>.<bucket_name>`
            prefixed_name = f"{owner_address}.{bucket_name}"
            return prefixed_name
        else:
            print(f"Could not find bucket with name: {bucket_name}")
            return None
    except (NoCredentialsError, PartialCredentialsError, Exception) as e:
        print(f"An error occurred while creating a bucket: {e}")
        return None

Interact with objects

The following functions interact with objects in the bucket, including writing, listing, and getting objects.

# Write an object to the store under the given key with a value
def write_object(client: S3Client, bucket: str, key: str, value: str) -> None:
    try:
        client.put_object(Bucket=bucket, Key=key, Body=value)
    except Exception as e:
        print(f"An error occurred while writing data: {e}")
 
 
# List objects in the store, optionally, with a prefix
def list_objects(
    client: S3Client, bucket: str, prefix: str | None = None
) -> list | None:
    options = {"Bucket": bucket}
    if prefix is not None:
        options["Prefix"] = prefix
    try:
        response = client.list_objects(**options)
        if "Contents" in response:
            objects = response["Contents"]
            return objects
        else:
            return None
    except Exception as e:
        print(f"An error occurred while listing objects: {e}")
        return None
 
 
# Get an object from the store by key
def get_object(client: S3Client, bucket: str, key: str) -> str | None:
    try:
        response = client.get_object(
            Bucket=bucket,
            Key=key,
        )
        return response["Body"].read()
    except Exception as e:
        print(f"An error occurred while getting object: {e}")
        return None

Connect, write, and read data

The example below demonstrates how to connect to the adapter, create a bucket, and interact with objects. It establishes a connection, creates a bucket, writes a simple key/value pair, and then reads it back.

The one callout is the OWNER_ADDRESS environment variable.

from os import getenv
from time import sleep
 
# Import the methods above if in a different file, or use accordingly
 
# Define your S3 adapter connection details
host = getenv("S3_HOST") or "http://localhost:8014"
access_key = getenv("S3_ACCESS_KEY") or "S3EXAMPLEAK"
secret_key = getenv("S3_SECRET_KEY") or "S3EXAMPLESK"
# Note: below is the public key of a localnet account with private key:
# 7c852118294e51e653712a81e05800f419141751be58f605c371e15141b007a6
owner_address = (
    getenv("OWNER_ADDRESS") or "0x90f79bf6eb2c4f870365e785982e1f101e93b906"
)
 
# Connect to the Recall S3 adapter
recall = connect_to_recall(host, access_key, secret_key)
if recall is None:
    raise RuntimeError("Failed to connect to Recall")
 
# Create a bucket
bucket = create_bucket(recall, "my-bucket", owner_address)
if bucket is None:
    raise RuntimeError("Failed to create bucket")
print(f"Bucket created successfully: {bucket}")
 
# Write an object under the `hello/world` key
key_prefix = "hello"
key = f"{key_prefix}/world"
write_object(recall, bucket, key, b"hello")
print(f"Uploaded file to Recall at key: {key}")
 
print("Waiting for transaction to settle...")
sleep(3)
 
# List objects
objects = list_objects(recall, bucket, f"{key_prefix}/")
if not objects:
    print("No objects found after upload")
    return
 
# Get an object from the list
object_key = objects[0]["Key"]
object_data = get_object(recall, bucket, object_key)
if object_data:
    print(f"Retrieved object: {object_data}")
else:
    print("Failed to retrieve object")