a06673b
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use bytes::Bytes;
use futures::TryStreamExt;
use http::Method;
use object_store::ObjectStoreExt;
use object_store::aws::{AmazonS3, AmazonS3Builder};
use object_store::path::Path;
use object_store::prefix::PrefixStore;
use object_store::signer::Signer;
use object_store::{GetResult, ObjectMeta, ObjectStore, PutResult};
use url::Url;
use crate::config::StorageSettings;
use crate::repo::{StorageError, StorageRepository};
#[derive(Debug)]
pub struct S3Storage {
store: Arc<PrefixStore<AmazonS3>>,
}
impl S3Storage {
pub fn from_config(settings: &StorageSettings) -> Result<Self, StorageError> {
let s3 = settings.get_s3();
let bucket = s3.get_bucket().clone();
let region = s3.get_region().clone();
if bucket.trim().is_empty() {
return Err(StorageError::Config(
"s3.bucket must not be empty".to_string(),
));
}
if region.trim().is_empty() {
return Err(StorageError::Config(
"s3.region must not be empty".to_string(),
));
}
let mut builder = AmazonS3Builder::new()
.with_bucket_name(bucket)
.with_region(region);
if let Some(endpoint) = s3.get_endpoint() {
builder = builder.with_endpoint(endpoint.clone());
}
if let Some(access_key_id) = s3.get_access_key_id() {
builder = builder.with_access_key_id(access_key_id.clone());
}
if let Some(secret_access_key) = s3.get_secret_access_key() {
builder = builder.with_secret_access_key(secret_access_key.clone());
}
if *s3.get_allow_http() {
builder = builder.with_allow_http(true);
}
let store = builder
.build()
.map_err(|e| StorageError::Config(format!("failed to initialize s3 storage: {e}")))?;
let store = match s3.get_prefix() {
Some(prefix) => PrefixStore::new(store, prefix.clone()),
None => PrefixStore::new(store, ""),
};
Ok(Self {
store: Arc::new(store),
})
}
}
#[async_trait]
impl StorageRepository for S3Storage {
async fn put(&self, path: &Path, data: Bytes) -> Result<PutResult, StorageError> {
Ok(self.store.put(path, data.into()).await?)
}
async fn get(&self, path: &Path) -> Result<GetResult, StorageError> {
Ok(self.store.get(path).await?)
}
async fn delete(&self, path: &Path) -> Result<(), StorageError> {
Ok(self.store.delete(path).await?)
}
async fn head(&self, path: &Path) -> Result<ObjectMeta, StorageError> {
Ok(self.store.head(path).await?)
}
async fn exists(&self, path: &Path) -> Result<bool, StorageError> {
Ok(self.store.head(path).await.is_ok())
}
async fn list(&self, prefix: Option<&Path>) -> Result<Vec<ObjectMeta>, StorageError> {
let stream = self.store.list(prefix);
Ok(stream.try_collect::<Vec<_>>().await?)
}
async fn copy(&self, from: &Path, to: &Path) -> Result<(), StorageError> {
Ok(self.store.copy(from, to).await?)
}
async fn presigned_upload(
&self,
path: &Path,
expires_in: Duration,
) -> Result<Url, StorageError> {
Ok(self.store.signed_url(Method::PUT, path, expires_in).await?)
}
async fn presigned_download(
&self,
path: &Path,
expires_in: Duration,
) -> Result<Url, StorageError> {
Ok(self.store.signed_url(Method::GET, path, expires_in).await?)
}
}