Skip to main content التثبيت باستخدام Codex أو Claude انسخ هذا Prompt والصقه في Codex أو Claude أو مساعد آخر ليراجع صفحة Skill ويثبّتها لك.
يتجاوز الأمر المباشر Prompt المخصّص للمراجعة. افحص المصدر قبل تشغيله.
npx skills add https://github.com/gar-ai/mallorn --skill rust-s3-patternsيبقى الأمر في سطر واحد. مرّر أفقيًا لمراجعته كاملًا قبل النسخ.
تفضّل نسخة محلية؟ نزّل الملفات المتاحة حاليًا لدى SkillsMP.
المهن ذات الصلةSOC
استنادا إلى تصنيف SOC المهني
| 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
[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,
) -> <> {
= Credentials::()
.(|e| Error::((, e)))?;
= (endpoint_url) = endpoint {
Region::Custom {
region: region.(),
endpoint: endpoint_url.(),
}
} {
region.()
.(|_| Region::Custom {
region: region.(),
endpoint: (, region),
})
};
= Bucket::(bucket_name, s3_region, credentials)?
.();
( {
bucket,
base_path: base_path.(),
})
}
(&, key: &) {
.base_path.() {
key.()
} {
(, .base_path, key)
}
}
}
Result
Self
let
credentials
from_env
map_err
S3
format!
"Failed to load credentials: {}"
let
s3_region
if
let
Some
to_string
to_string
else
parse
unwrap_or_else
to_string
format!
"https://s3.{}.amazonaws.com"
let
bucket
new
with_path_style
Ok
Self
to_string
fn
full_key
self
str
->
String
if
self
is_empty
to_string
else
format!
"{}/{}"
self
Upload Operations
impl StorageClient {
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))
}
pub async fn upload_local_file(
&self,
local_path: &Path,
s3_key: &str,
content_type: Option<&str>,
) -> Result<String> {
let data = tokio::fs::read(local_path).await?;
self.upload_file(s3_key, Bytes::from(data), content_type).await
}
pub async fn upload_auto(
&self,
local_path: &Path,
s3_key: &str,
) -> Result<String> {
let content_type = mime_guess::from_path(local_path)
.first_raw()
.unwrap_or("application/octet-stream");
self.upload_local_file(local_path, s3_key, Some(content_type)).await
}
}
Download Operations
impl StorageClient {
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()))
}
pub async fn download_to_file(
&self,
key: &str,
local_path: &Path,
) -> Result<()> {
let data = self.download_file(key).await?;
if let Some(parent) = local_path.parent() {
tokio::fs::create_dir_all(parent).await?;
}
tokio::fs::write(local_path, &data).await?;
Ok(())
}
pub async fn download_streaming(
&self,
key: &str,
writer: &mut (impl tokio::io::AsyncWrite + Unpin),
) -> Result<u64> {
let full_key = self.full_key(key);
let response = self.bucket
.get_object_stream(&full_key)
.await
.map_err(|e| Error::S3(format!("Stream download failed: {}", e)))?;
let bytes_written = tokio::io::copy(&mut response.bytes.as_ref(), writer).await?;
Ok(bytes_written)
}
}
Existence and Metadata
impl StorageClient {
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)))
}
}
}
}
pub async fn get_metadata(&self, key: &str) -> Result<ObjectMetadata> {
let full_key = self.full_key(key);
let (head, _) = self.bucket
.head_object(&full_key)
.await
.map_err(|e| Error::S3(format!("Head object failed: {}", e)))?;
Ok(ObjectMetadata {
size: head.content_length.unwrap_or(0) as u64,
content_type: head.content_type,
last_modified: head.last_modified,
})
}
}
Deletion
impl StorageClient {
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(())
}
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 {
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)
}
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 as u32, None)
.await
.map_err(|e| Error::S3(format!("Presign failed: {}", e)))?;
Ok(url)
}
}
Batch Operations with Semaphore
use std::sync::Arc;
use tokio::sync::Semaphore;
use futures::stream::{self, StreamExt};
impl StorageClient {
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).await?;
Ok::<PathBuf, Error>(local_path)
}
})
.buffer_unordered(max_concurrent * 2)
.collect()
.await;
results.into_iter().collect()
}
pub async fn upload_directory(
&self,
local_dir: &Path,
s3_prefix: &str,
max_concurrent: usize,
) -> Result<usize> {
let semaphore = Arc::new(Semaphore::new(max_concurrent));
let mut files = Vec::new();
let mut entries = tokio::fs::read_dir(local_dir).await?;
while let Some(entry) = entries.next_entry().await? {
if entry.file_type().await?.is_file() {
files.push(entry.path());
}
}
let results: Vec<_> = stream::iter(files)
.map(|path| {
let sem = semaphore.clone();
let client = self.clone();
let file_name = path.file_name().unwrap().to_string_lossy().to_string();
let s3_key = format!("{}/{}", s3_prefix, file_name);
async move {
let _permit = sem.acquire().await?;
client.upload_local_file(&path, &s3_key, None).await
}
})
.buffer_unordered(max_concurrent * 2)
.collect()
.await;
let uploaded = results.iter().filter(|r| r.is_ok()).count();
Ok(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.