use crate::{
multipart::{MultipartStore, PartId},
path::Path,
signer::Signer,
GetOptions, GetResult, ListResult, MultipartId, MultipartUpload, ObjectMeta, ObjectStore,
PutMultipartOpts, PutOptions, PutPayload, PutResult, Result, UploadPart,
};
use async_trait::async_trait;
use futures::stream::BoxStream;
use reqwest::Method;
use std::fmt::Debug;
use std::sync::Arc;
use std::time::Duration;
use url::Url;
use crate::client::get::GetClientExt;
use crate::client::list::ListClientExt;
use crate::client::CredentialProvider;
pub use credential::{authority_hosts, AzureAccessKey, AzureAuthorizer};
mod builder;
mod client;
mod credential;
pub type AzureCredentialProvider = Arc<dyn CredentialProvider<Credential = AzureCredential>>;
use crate::azure::client::AzureClient;
use crate::client::parts::Parts;
pub use builder::{AzureConfigKey, MicrosoftAzureBuilder};
pub use credential::AzureCredential;
const STORE: &str = "MicrosoftAzure";
#[derive(Debug)]
pub struct MicrosoftAzure {
client: Arc<client::AzureClient>,
}
impl MicrosoftAzure {
pub fn credentials(&self) -> &AzureCredentialProvider {
&self.client.config().credentials
}
fn path_url(&self, path: &Path) -> url::Url {
self.client.config().path_url(path)
}
}
impl std::fmt::Display for MicrosoftAzure {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"MicrosoftAzure {{ account: {}, container: {} }}",
self.client.config().account,
self.client.config().container
)
}
}
#[async_trait]
impl ObjectStore for MicrosoftAzure {
async fn put_opts(
&self,
location: &Path,
payload: PutPayload,
opts: PutOptions,
) -> Result<PutResult> {
self.client.put_blob(location, payload, opts).await
}
async fn put_multipart_opts(
&self,
location: &Path,
opts: PutMultipartOpts,
) -> Result<Box<dyn MultipartUpload>> {
Ok(Box::new(AzureMultiPartUpload {
part_idx: 0,
opts,
state: Arc::new(UploadState {
client: Arc::clone(&self.client),
location: location.clone(),
parts: Default::default(),
}),
}))
}
async fn get_opts(&self, location: &Path, options: GetOptions) -> Result<GetResult> {
self.client.get_opts(location, options).await
}
async fn delete(&self, location: &Path) -> Result<()> {
self.client.delete_request(location, &()).await
}
fn list(&self, prefix: Option<&Path>) -> BoxStream<'_, Result<ObjectMeta>> {
self.client.list(prefix)
}
async fn list_with_delimiter(&self, prefix: Option<&Path>) -> Result<ListResult> {
self.client.list_with_delimiter(prefix).await
}
async fn copy(&self, from: &Path, to: &Path) -> Result<()> {
self.client.copy_request(from, to, true).await
}
async fn copy_if_not_exists(&self, from: &Path, to: &Path) -> Result<()> {
self.client.copy_request(from, to, false).await
}
}
#[async_trait]
impl Signer for MicrosoftAzure {
async fn signed_url(&self, method: Method, path: &Path, expires_in: Duration) -> Result<Url> {
let mut url = self.path_url(path);
let signer = self.client.signer(expires_in).await?;
signer.sign(&method, &mut url)?;
Ok(url)
}
async fn signed_urls(
&self,
method: Method,
paths: &[Path],
expires_in: Duration,
) -> Result<Vec<Url>> {
let mut urls = Vec::with_capacity(paths.len());
let signer = self.client.signer(expires_in).await?;
for path in paths {
let mut url = self.path_url(path);
signer.sign(&method, &mut url)?;
urls.push(url);
}
Ok(urls)
}
}
#[derive(Debug)]
struct AzureMultiPartUpload {
part_idx: usize,
state: Arc<UploadState>,
opts: PutMultipartOpts,
}
#[derive(Debug)]
struct UploadState {
location: Path,
parts: Parts,
client: Arc<AzureClient>,
}
#[async_trait]
impl MultipartUpload for AzureMultiPartUpload {
fn put_part(&mut self, data: PutPayload) -> UploadPart {
let idx = self.part_idx;
self.part_idx += 1;
let state = Arc::clone(&self.state);
Box::pin(async move {
let part = state.client.put_block(&state.location, idx, data).await?;
state.parts.put(idx, part);
Ok(())
})
}
async fn complete(&mut self) -> Result<PutResult> {
let parts = self.state.parts.finish(self.part_idx)?;
self.state
.client
.put_block_list(&self.state.location, parts, std::mem::take(&mut self.opts))
.await
}
async fn abort(&mut self) -> Result<()> {
Ok(())
}
}
#[async_trait]
impl MultipartStore for MicrosoftAzure {
async fn create_multipart(&self, _: &Path) -> Result<MultipartId> {
Ok(String::new())
}
async fn put_part(
&self,
path: &Path,
_: &MultipartId,
part_idx: usize,
data: PutPayload,
) -> Result<PartId> {
self.client.put_block(path, part_idx, data).await
}
async fn complete_multipart(
&self,
path: &Path,
_: &MultipartId,
parts: Vec<PartId>,
) -> Result<PutResult> {
self.client
.put_block_list(path, parts, Default::default())
.await
}
async fn abort_multipart(&self, _: &Path, _: &MultipartId) -> Result<()> {
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::integration::*;
use crate::tests::*;
use bytes::Bytes;
#[tokio::test]
async fn azure_blob_test() {
maybe_skip_integration!();
let integration = MicrosoftAzureBuilder::from_env().build().unwrap();
put_get_delete_list(&integration).await;
get_opts(&integration).await;
list_uses_directories_correctly(&integration).await;
list_with_delimiter(&integration).await;
rename_and_copy(&integration).await;
copy_if_not_exists(&integration).await;
stream_get(&integration).await;
put_opts(&integration, true).await;
multipart(&integration, &integration).await;
signing(&integration).await;
let validate = !integration.client.config().disable_tagging;
tagging(
Arc::new(MicrosoftAzure {
client: Arc::clone(&integration.client),
}),
validate,
|p| {
let client = Arc::clone(&integration.client);
async move { client.get_blob_tagging(&p).await }
},
)
.await;
if !integration.client.config().is_emulator {
put_get_attributes(&integration).await;
}
}
#[ignore = "Used for manual testing against a real storage account."]
#[tokio::test]
async fn test_user_delegation_key() {
let account = std::env::var("AZURE_ACCOUNT_NAME").unwrap();
let container = std::env::var("AZURE_CONTAINER_NAME").unwrap();
let client_id = std::env::var("AZURE_CLIENT_ID").unwrap();
let client_secret = std::env::var("AZURE_CLIENT_SECRET").unwrap();
let tenant_id = std::env::var("AZURE_TENANT_ID").unwrap();
let integration = MicrosoftAzureBuilder::new()
.with_account(account)
.with_container_name(container)
.with_client_id(client_id)
.with_client_secret(client_secret)
.with_tenant_id(&tenant_id)
.build()
.unwrap();
let data = Bytes::from("hello world");
let path = Path::from("file.txt");
integration.put(&path, data.clone().into()).await.unwrap();
let signed = integration
.signed_url(Method::GET, &path, Duration::from_secs(60))
.await
.unwrap();
let resp = reqwest::get(signed).await.unwrap();
let loaded = resp.bytes().await.unwrap();
assert_eq!(data, loaded);
}
#[test]
fn azure_test_config_get_value() {
let azure_client_id = "object_store:fake_access_key_id".to_string();
let azure_storage_account_name = "object_store:fake_secret_key".to_string();
let azure_storage_token = "object_store:fake_default_region".to_string();
let builder = MicrosoftAzureBuilder::new()
.with_config(AzureConfigKey::ClientId, &azure_client_id)
.with_config(AzureConfigKey::AccountName, &azure_storage_account_name)
.with_config(AzureConfigKey::Token, &azure_storage_token);
assert_eq!(
builder.get_config_value(&AzureConfigKey::ClientId).unwrap(),
azure_client_id
);
assert_eq!(
builder
.get_config_value(&AzureConfigKey::AccountName)
.unwrap(),
azure_storage_account_name
);
assert_eq!(
builder.get_config_value(&AzureConfigKey::Token).unwrap(),
azure_storage_token
);
}
}