Agent Object Storage

This document provides guidance on how to use the ObjectStoreAPI class to interact with an S3-compatible object storage backend (such as AWS S3, MinIO, Ceph, etc.). The class supports uploading and downloading files using various data formats (file path, bytes, or in-memory buffers).

---

Introduction

The ObjectStoreAPI class provides a simple and extensible interface to:

• Upload files to an S3-compatible bucket.
• Download files from a bucket to a path, a buffer, or directly as bytes.

It supports:

• Credentials and region-based authentication.
• Multiple input and output types.
• Logging for success and error cases.

This can be extended or used with AWS S3, or with custom object storage systems that implement the S3 API (e.g., MinIO).

---

Import

from utils.object_store import ObjectStoreAPI

---

Usage

Configuration

config = {
    "aws_access_key_id": "your-access-key",
    "aws_secret_access_key": "your-secret-key",
    "region_name": "your-region"
}

Initialize the client

store = ObjectStoreAPI(config)

Upload file from file path

store.upload_file(bucket_name="my-bucket", object_key="my-folder/my-file.txt", data="/path/to/local/file.txt")

Upload file from bytes

binary_data = b"Hello, world!"
store.upload_file(bucket_name="my-bucket", object_key="data/hello.txt", data=binary_data)

Upload file from a BytesIO buffer

import io

buffer = io.BytesIO()
buffer.write(b"Buffered content")
buffer.seek(0)
store.upload_file(bucket_name="my-bucket", object_key="buffer/file.txt", data=buffer)

---

Download file to local path

store.download_file(bucket_name="my-bucket", object_key="my-folder/my-file.txt", destination="/path/to/save/file.txt")

Download file to a BytesIO buffer

import io

output_buffer = io.BytesIO()
store.download_file(bucket_name="my-bucket", object_key="data/hello.txt", destination=output_buffer)
print(output_buffer.getvalue())

Download file and get content as bytes

data = store.download_file(bucket_name="my-bucket", object_key="data/hello.txt")
print(data.decode())

---

Custom Object Storage

To use a custom object storage system like MinIO or Ceph:

1. Ensure it supports the S3 API.
2. Replace the configuration:

config = {
    "aws_access_key_id": "your-key",
    "aws_secret_access_key": "your-secret",
    "region_name": "us-east-1"
}

# If using a custom endpoint:
import boto3

store = ObjectStoreAPI(config)
store.s3_client = boto3.client(
    's3',
    endpoint_url="https://your-custom-object-store.local",
    aws_access_key_id=config["aws_access_key_id"],
    aws_secret_access_key=config["aws_secret_access_key"],
    region_name=config["region_name"]
)

This allows you to seamlessly use the same ObjectStoreAPI with private or self-hosted storage systems.