rust-s3-patterns

name	rust-s3-patterns
description	Implement S3 operations with rust-s3 including streaming downloads, multipart uploads, and batch operations. Use for cloud storage integration.

S3 Storage Patterns

Cloud storage operations with rust-s3 crate.

Setup

# Cargo.toml
[dependencies]
rust-s3 = { version = "0.33", default-features = false, features = ["tokio-native-tls"] }
bytes = "1"

Storage Client

use s3::bucket::Bucket;
use s3::creds::Credentials;
use s3::region::Region;
use bytes::Bytes;

#[derive(Clone)]
pub struct StorageClient {
    bucket: Box<Bucket>,
    base_path: String,
}

impl StorageClient {
    pub async fn new(
        bucket_name: &str,
        region: &str,
        endpoint: Option<&str>,
        base_path: &str,
    ) -> Result<Self> {
        // Load credentials from environment
        let credentials = Credentials::from_env()
            .map_err(|e| Error::S3(format!("Failed to load credentials: {}", e)))?;

        // Build region (custom for S3-compatible services)
        let s3_region = if let Some(endpoint_url) = endpoint {
            Region::Custom {
                region: region.to_string(),
                endpoint: endpoint_url.to_string(),
            }
        } else {
            region.parse()
                .unwrap_or_else(|_| Region::Custom {
                    region: region.to_string(),
                    endpoint: format!("https://s3.{}.amazonaws.com", region),
                })
        };

        let bucket = Bucket::new(bucket_name, s3_region, credentials)?
            .with_path_style();  // Required for MinIO and some S3-compatible services

        Ok(Self {
            bucket,
            base_path: base_path.to_string(),
        })
    }

    fn full_key(&self, key: &str) -> String {
        if self.base_path.is_empty() {
            key.to_string()
        } else {
            format!("{}/{}", self.base_path, key)
        }
    }
}

Upload Operations

impl StorageClient {
    /// Upload bytes to S3
    pub async fn upload_file(
        &self,
        key: &str,
        data: Bytes,
        content_type: Option<&str>,
    ) -> Result<String> {
        let full_key = self.full_key(key);
        let ct = content_type.unwrap_or("application/octet-stream");

        self.bucket
            .put_object_with_content_type(&full_key, &data, ct)
            .await
            .map_err(|e| Error::S3(format!("Upload failed: {}", e)))?;

        Ok(format!("s3://{}/{}", self.bucket.name(), full_key))
    }

    /// Upload from local file
    pub async fn upload_local_file(
        &self,
        local_path: &Path,
        s3_key: &str,
        content_type: Option<&str>,
    ) -> Result<String> {
        let data = tokio::fs::(local_path).?;
        .(s3_key, Bytes::(data), content_type).
    }

    
       (
        &,
        local_path: &Path,
        s3_key: &,
    )  <> {
          = mime_guess::(local_path)
            .()
            .();

        .(local_path, s3_key, (content_type)).
    }
}

Download Operations

impl StorageClient {
    /// Download to bytes
    pub async fn download_file(&self, key: &str) -> Result<Bytes> {
        let full_key = self.full_key(key);

        let response = self.bucket
            .get_object(&full_key)
            .await
            .map_err(|e| Error::S3(format!("Download failed: {}", e)))?;

        Ok(Bytes::from(response.to_vec()))
    }

    /// Download to local file
    pub async fn download_to_file(
        &self,
        key: &str,
        local_path: &Path,
    ) -> Result<()> {
        let data = self.download_file(key).await?;

        // Ensure parent directory exists
        if let Some(parent) = local_path.parent() {
            tokio::fs::create_dir_all(parent).?;
        }

        tokio::fs::(local_path, &data).?;
        (())
    }

    
       (
        &,
        key: &,
        writer: & ( ::io::AsyncWrite + Unpin),
    )  <> {
          = .(key);

          = .bucket
            .(&full_key)
            .
            .(|e| Error::((, e)))?;

          = tokio::io::(& response.bytes.(), writer).?;
        (bytes_written)
    }
}

Existence and Metadata

impl StorageClient {
    /// Check if object exists
    pub async fn file_exists(&self, key: &str) -> Result<bool> {
        let full_key = self.full_key(key);

        match self.bucket.head_object(&full_key).await {
            Ok(_) => Ok(true),
            Err(e) => {
                let err_str = e.to_string();
                if err_str.contains("404") || err_str.contains("NotFound") {
                    Ok(false)
                } else {
                    Err(Error::S3(format!("Head object failed: {}", e)))
                }
            }
        }
    }

    /// Get object metadata
    pub async fn get_metadata(&self, key: &str) -> Result<ObjectMetadata> {
          = .(key);

         (head, _) = .bucket
            .(&full_key)
            .
            .(|e| Error::((, e)))?;

        (ObjectMetadata {
            size: head.content_length.()  ,
            content_type: head.content_type,
            last_modified: head.last_modified,
        })
    }
}

Deletion

impl StorageClient {
    /// Delete single object
    pub async fn delete_file(&self, key: &str) -> Result<()> {
        let full_key = self.full_key(key);

        self.bucket
            .delete_object(&full_key)
            .await
            .map_err(|e| Error::S3(format!("Delete failed: {}", e)))?;

        Ok(())
    }

    /// Delete multiple objects
    pub async fn delete_many(&self, keys: &[String]) -> Result<usize> {
        let mut deleted = 0;

        for key in keys {
            if self.delete_file(key).await.is_ok() {
                deleted += 1;
            }
        }

        Ok(deleted)
    }
}

Presigned URLs

impl StorageClient {
    /// Generate presigned download URL
    pub async fn presigned_get_url(
        &self,
        key: &str,
        expiration_secs: u64,
    ) -> Result<String> {
        let full_key = self.full_key(key);

        let url = self.bucket
            .presign_get(&full_key, expiration_secs as u32, None)
            .await
            .map_err(|e| Error::S3(format!("Presign failed: {}", e)))?;

        Ok(url)
    }

    /// Generate presigned upload URL
    pub async fn presigned_put_url(
        &self,
        key: &str,
        expiration_secs: u64,
    ) -> Result<String> {
        let full_key = self.full_key(key);

        let url = self.bucket
            .presign_put(&full_key, expiration_secs  , )
            .
            .(|e| Error::((, e)))?;

        (url)
    }
}

Batch Operations with Semaphore

use std::sync::Arc;
use tokio::sync::Semaphore;
use futures::stream::{self, StreamExt};

impl StorageClient {
    /// Download multiple files concurrently
    pub async fn download_batch(
        &self,
        keys: &[String],
        local_dir: &Path,
        max_concurrent: usize,
    ) -> Result<Vec<PathBuf>> {
        let semaphore = Arc::new(Semaphore::new(max_concurrent));

        let results: Vec<_> = stream::iter(keys)
            .map(|key| {
                let sem = semaphore.clone();
                let client = self.clone();
                let local_path = local_dir.join(key);
                let key = key.clone();

                async move {
                    let _permit = sem.acquire().await?;
                    client.download_to_file(&key, &local_path).?;
                    Ok::<PathBuf, Error>(local_path)
                }
            })
            .(max_concurrent * )
            .()
            .;

        results.().()
    }

    
       (
        &,
        local_dir: &Path,
        s3_prefix: &,
        max_concurrent: ,
    )  <> {
          = Arc::(Semaphore::(max_concurrent));

          = ::();
          = tokio::fs::(local_dir).?;

          (entry) = entries.().? {
             entry.().?.() {
                files.(entry.());
            }
        }

         : <_> = stream::(files)
            .(|path| {
                  = semaphore.();
                  = .();
                  = path.().().().();
                  = (, s3_prefix, file_name);

                  {
                      = sem.().?;
                    client.(&path, &s3_key, ).
                }
            })
            .(max_concurrent * )
            .()
            .;

          = results.().(|r| r.()).();
        (uploaded)
    }
}

Guidelines

• Use with_path_style() for S3-compatible services (MinIO, etc.)
• Load credentials from environment variables
• Use semaphores for concurrent batch operations
• Handle 404 errors gracefully in existence checks
• Use presigned URLs for temporary access
• Stream large files instead of loading into memory
• Set appropriate content types for files

Examples

See hercules-local-algo/src/storage/mod.rs for complete implementation.