Add an S3 Bucket
Step-by-step guide to adding and configuring an S3 bucket in your Shuttle Cobra project.
Prerequisites
Instructions: Provisioning an S3 Bucket
1. Install Dependencies
uv init
uv add shuttle-cobra2. Define Your S3 Bucket in main.py
main.pyfrom typing import Annotated
import shuttle_task
import shuttle_runtime
from shuttle_aws.s3 import Bucket, BucketOptions
# Define a constant for your desired bucket name
BUCKET_NAME = "my-shuttle-cobra-bucket"
@shuttle_task.cron(schedule="0 * * * ? *") # This task will run hourly
async def main(
bucket: Annotated[
Bucket,
BucketOptions(bucket_name=BUCKET_NAME, policies=[])
],
):
"""An example task that interacts with an S3 bucket."""
print(f"Accessing S3 bucket: {bucket.options.bucket_name}")
# Get a boto3 S3 client to interact with the bucket
s3_client = bucket.get_client()
try:
# Example: List objects in the bucket
response = s3_client.list_objects_v2(Bucket=BUCKET_NAME)
if "Contents" in response:
print(f"Found {response['KeyCount']} objects in '{BUCKET_NAME}':")
for obj in response["Contents"]:
print(f"- {obj['Key']} ({obj['Size']} bytes)")
else:
print(f"No objects found in '{BUCKET_NAME}'.")
# Example: Put a simple object
s3_client.put_object(Bucket=BUCKET_NAME, Key="hello.txt", Body="Hello from Shuttle Cobra!")
print("Successfully put 'hello.txt' into the bucket.")
except Exception as e:
print(f"Error interacting with S3 bucket: {e}")
# This line is essential for Shuttle to run your application
if __name__ == "__main__":
shuttle_runtime.main(main)3. Add External Write Permissions (Optional)
4. Deploy to the Cloud
5. Test Locally
Troubleshooting
Next Steps
Last updated