opendal_core/services/azblob/
core.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::fmt::Debug;
19use std::sync::Arc;
20
21use base64::Engine;
22use base64::prelude::BASE64_STANDARD;
23use bytes::Bytes;
24use constants::X_MS_META_PREFIX;
25use http::HeaderValue;
26use http::Request;
27use http::Response;
28use http::header::CONTENT_LENGTH;
29use http::header::CONTENT_TYPE;
30use http::header::HeaderName;
31use http::header::IF_MATCH;
32use http::header::IF_MODIFIED_SINCE;
33use http::header::IF_NONE_MATCH;
34use http::header::IF_UNMODIFIED_SINCE;
35use reqsign::AzureStorageCredential;
36use reqsign::AzureStorageLoader;
37use reqsign::AzureStorageSigner;
38use serde::Deserialize;
39use serde::Serialize;
40use uuid::Uuid;
41
42use crate::raw::*;
43use crate::*;
44
45pub mod constants {
46    // Indicates the Blob Storage version that was used to execute the request
47    pub const X_MS_VERSION: &str = "x-ms-version";
48
49    pub const X_MS_BLOB_TYPE: &str = "x-ms-blob-type";
50    pub const X_MS_COPY_SOURCE: &str = "x-ms-copy-source";
51    pub const X_MS_BLOB_CACHE_CONTROL: &str = "x-ms-blob-cache-control";
52    pub const X_MS_BLOB_CONDITION_APPENDPOS: &str = "x-ms-blob-condition-appendpos";
53    pub const X_MS_META_PREFIX: &str = "x-ms-meta-";
54
55    // indicates the version of the blob, and it can be used in subsequent requests to access the blob.
56    pub const X_MS_VERSION_ID: &str = "x-ms-version-id";
57
58    // Server-side encryption with customer-provided headers
59    pub const X_MS_ENCRYPTION_KEY: &str = "x-ms-encryption-key";
60    pub const X_MS_ENCRYPTION_KEY_SHA256: &str = "x-ms-encryption-key-sha256";
61    pub const X_MS_ENCRYPTION_ALGORITHM: &str = "x-ms-encryption-algorithm";
62}
63
64pub struct AzblobCore {
65    pub info: Arc<AccessorInfo>,
66    pub container: String,
67    pub root: String,
68    pub endpoint: String,
69    pub encryption_key: Option<HeaderValue>,
70    pub encryption_key_sha256: Option<HeaderValue>,
71    pub encryption_algorithm: Option<HeaderValue>,
72    pub loader: AzureStorageLoader,
73    pub signer: AzureStorageSigner,
74}
75
76impl Debug for AzblobCore {
77    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
78        f.debug_struct("AzblobCore")
79            .field("container", &self.container)
80            .field("root", &self.root)
81            .field("endpoint", &self.endpoint)
82            .finish_non_exhaustive()
83    }
84}
85
86impl AzblobCore {
87    async fn load_credential(&self) -> Result<AzureStorageCredential> {
88        let cred = self
89            .loader
90            .load()
91            .await
92            .map_err(new_request_credential_error)?;
93
94        if let Some(cred) = cred {
95            Ok(cred)
96        } else {
97            Err(Error::new(
98                ErrorKind::ConfigInvalid,
99                "no valid credential found",
100            ))
101        }
102    }
103
104    pub async fn sign_query<T>(&self, req: &mut Request<T>) -> Result<()> {
105        let cred = self.load_credential().await?;
106
107        self.signer
108            .sign_query(req, Duration::from_secs(3600), &cred)
109            .map_err(new_request_sign_error)
110    }
111
112    pub async fn sign<T>(&self, req: &mut Request<T>) -> Result<()> {
113        let cred = self.load_credential().await?;
114        // Insert x-ms-version header for normal requests.
115        req.headers_mut().insert(
116            HeaderName::from_static(constants::X_MS_VERSION),
117            // 2022-11-02 is the version supported by Azurite V3 and
118            // used by Azure Portal, We use this version to make
119            // sure most our developer happy.
120            //
121            // In the future, we could allow users to configure this value.
122            HeaderValue::from_static("2022-11-02"),
123        );
124        self.signer.sign(req, &cred).map_err(new_request_sign_error)
125    }
126
127    async fn batch_sign<T>(&self, req: &mut Request<T>) -> Result<()> {
128        let cred = self.load_credential().await?;
129        self.signer.sign(req, &cred).map_err(new_request_sign_error)
130    }
131
132    #[inline]
133    pub async fn send(&self, req: Request<Buffer>) -> Result<Response<Buffer>> {
134        self.info.http_client().send(req).await
135    }
136
137    pub fn insert_sse_headers(&self, mut req: http::request::Builder) -> http::request::Builder {
138        if let Some(v) = &self.encryption_key {
139            let mut v = v.clone();
140            v.set_sensitive(true);
141
142            req = req.header(HeaderName::from_static(constants::X_MS_ENCRYPTION_KEY), v)
143        }
144
145        if let Some(v) = &self.encryption_key_sha256 {
146            let mut v = v.clone();
147            v.set_sensitive(true);
148
149            req = req.header(
150                HeaderName::from_static(constants::X_MS_ENCRYPTION_KEY_SHA256),
151                v,
152            )
153        }
154
155        if let Some(v) = &self.encryption_algorithm {
156            let mut v = v.clone();
157            v.set_sensitive(true);
158
159            req = req.header(
160                HeaderName::from_static(constants::X_MS_ENCRYPTION_ALGORITHM),
161                v,
162            )
163        }
164
165        req
166    }
167}
168
169impl AzblobCore {
170    fn build_path_url(&self, path: &str) -> String {
171        format!(
172            "{}/{}/{}",
173            self.endpoint,
174            self.container,
175            percent_encode_path(&build_abs_path(&self.root, path))
176        )
177    }
178
179    pub fn azblob_get_blob_request(
180        &self,
181        path: &str,
182        range: BytesRange,
183        args: &OpRead,
184    ) -> Result<Request<Buffer>> {
185        let mut url = self.build_path_url(path);
186
187        if let Some(override_content_disposition) = args.override_content_disposition() {
188            url.push_str(&format!(
189                "?rscd={}",
190                percent_encode_path(override_content_disposition)
191            ));
192        }
193
194        let mut req = Request::get(&url);
195
196        // Set SSE headers.
197        req = self.insert_sse_headers(req);
198
199        if !range.is_full() {
200            req = req.header(http::header::RANGE, range.to_header());
201        }
202
203        if let Some(if_none_match) = args.if_none_match() {
204            req = req.header(IF_NONE_MATCH, if_none_match);
205        }
206
207        if let Some(if_match) = args.if_match() {
208            req = req.header(IF_MATCH, if_match);
209        }
210
211        if let Some(if_modified_since) = args.if_modified_since() {
212            req = req.header(IF_MODIFIED_SINCE, if_modified_since.format_http_date());
213        }
214
215        if let Some(if_unmodified_since) = args.if_unmodified_since() {
216            req = req.header(IF_UNMODIFIED_SINCE, if_unmodified_since.format_http_date());
217        }
218
219        let req = req
220            .extension(Operation::Read)
221            .body(Buffer::new())
222            .map_err(new_request_build_error)?;
223
224        Ok(req)
225    }
226
227    pub async fn azblob_get_blob(
228        &self,
229        path: &str,
230        range: BytesRange,
231        args: &OpRead,
232    ) -> Result<Response<HttpBody>> {
233        let mut req = self.azblob_get_blob_request(path, range, args)?;
234
235        self.sign(&mut req).await?;
236
237        self.info.http_client().fetch(req).await
238    }
239
240    pub fn azblob_put_blob_request(
241        &self,
242        path: &str,
243        size: Option<u64>,
244        args: &OpWrite,
245        body: Buffer,
246    ) -> Result<Request<Buffer>> {
247        let mut req = Request::put(self.build_path_url(path));
248
249        req = req.header(
250            HeaderName::from_static(constants::X_MS_BLOB_TYPE),
251            "BlockBlob",
252        );
253
254        if let Some(size) = size {
255            req = req.header(CONTENT_LENGTH, size)
256        }
257
258        if let Some(ty) = args.content_type() {
259            req = req.header(CONTENT_TYPE, ty)
260        }
261
262        // Specify the wildcard character (*) to perform the operation only if
263        // the resource does not exist, and fail the operation if it does exist.
264        if args.if_not_exists() {
265            req = req.header(IF_NONE_MATCH, "*");
266        }
267
268        if let Some(v) = args.if_none_match() {
269            req = req.header(IF_NONE_MATCH, v);
270        }
271
272        if let Some(cache_control) = args.cache_control() {
273            req = req.header(constants::X_MS_BLOB_CACHE_CONTROL, cache_control);
274        }
275
276        // Set SSE headers.
277        req = self.insert_sse_headers(req);
278
279        if let Some(user_metadata) = args.user_metadata() {
280            for (key, value) in user_metadata {
281                req = req.header(format!("{X_MS_META_PREFIX}{key}"), value)
282            }
283        }
284
285        let req = req
286            .extension(Operation::Write)
287            .body(body)
288            .map_err(new_request_build_error)?;
289
290        Ok(req)
291    }
292
293    pub async fn azblob_put_blob(
294        &self,
295        path: &str,
296        size: Option<u64>,
297        args: &OpWrite,
298        body: Buffer,
299    ) -> Result<Response<Buffer>> {
300        let mut req = self.azblob_put_blob_request(path, size, args, body)?;
301
302        self.sign(&mut req).await?;
303        self.send(req).await
304    }
305
306    /// For appendable object, it could be created by `put` an empty blob
307    /// with `x-ms-blob-type` header set to `AppendBlob`.
308    /// And it's just initialized with empty content.
309    ///
310    /// If want to append content to it, we should use the following method `azblob_append_blob_request`.
311    ///
312    /// # Notes
313    ///
314    /// Appendable blob's custom header could only be set when it's created.
315    ///
316    /// The following custom header could be set:
317    /// - `content-type`
318    /// - `x-ms-blob-cache-control`
319    ///
320    /// # Reference
321    ///
322    /// https://learn.microsoft.com/en-us/rest/api/storageservices/put-blob
323    fn azblob_init_appendable_blob_request(
324        &self,
325        path: &str,
326        args: &OpWrite,
327    ) -> Result<Request<Buffer>> {
328        let mut req = Request::put(self.build_path_url(path));
329
330        // Set SSE headers.
331        req = self.insert_sse_headers(req);
332
333        // The content-length header must be set to zero
334        // when creating an appendable blob.
335        req = req.header(CONTENT_LENGTH, 0);
336        req = req.header(
337            HeaderName::from_static(constants::X_MS_BLOB_TYPE),
338            "AppendBlob",
339        );
340
341        if let Some(ty) = args.content_type() {
342            req = req.header(CONTENT_TYPE, ty)
343        }
344
345        if let Some(cache_control) = args.cache_control() {
346            req = req.header(constants::X_MS_BLOB_CACHE_CONTROL, cache_control);
347        }
348
349        let req = req
350            .extension(Operation::Write)
351            .body(Buffer::new())
352            .map_err(new_request_build_error)?;
353
354        Ok(req)
355    }
356
357    pub async fn azblob_init_appendable_blob(
358        &self,
359        path: &str,
360        args: &OpWrite,
361    ) -> Result<Response<Buffer>> {
362        let mut req = self.azblob_init_appendable_blob_request(path, args)?;
363
364        self.sign(&mut req).await?;
365        self.send(req).await
366    }
367
368    /// Append content to an appendable blob.
369    /// The content will be appended to the end of the blob.
370    ///
371    /// # Notes
372    ///
373    /// - The maximum size of the content could be appended is 4MB.
374    /// - `Append Block` succeeds only if the blob already exists.
375    ///
376    /// # Reference
377    ///
378    /// https://learn.microsoft.com/en-us/rest/api/storageservices/append-block
379    fn azblob_append_blob_request(
380        &self,
381        path: &str,
382        position: u64,
383        size: u64,
384        body: Buffer,
385    ) -> Result<Request<Buffer>> {
386        let url = format!("{}?comp=appendblock", &self.build_path_url(path));
387
388        let mut req = Request::put(&url)
389            .header(CONTENT_LENGTH, size)
390            .header(constants::X_MS_BLOB_CONDITION_APPENDPOS, position);
391
392        // Set SSE headers.
393        req = self.insert_sse_headers(req);
394
395        let req = req
396            .extension(Operation::Write)
397            .body(body)
398            .map_err(new_request_build_error)?;
399
400        Ok(req)
401    }
402
403    pub async fn azblob_append_blob(
404        &self,
405        path: &str,
406        position: u64,
407        size: u64,
408        body: Buffer,
409    ) -> Result<Response<Buffer>> {
410        let mut req = self.azblob_append_blob_request(path, position, size, body)?;
411
412        self.sign(&mut req).await?;
413        self.send(req).await
414    }
415
416    pub fn azblob_put_block_request(
417        &self,
418        path: &str,
419        block_id: Uuid,
420        size: Option<u64>,
421        args: &OpWrite,
422        body: Buffer,
423    ) -> Result<Request<Buffer>> {
424        // To be written as part of a blob, a block must have been successfully written to the server in an earlier Put Block operation.
425        // refer to https://learn.microsoft.com/en-us/rest/api/storageservices/put-block?tabs=microsoft-entra-id
426        let url = QueryPairsWriter::new(&self.build_path_url(path))
427            .push("comp", "block")
428            .push(
429                "blockid",
430                &percent_encode_path(&BASE64_STANDARD.encode(block_id.as_bytes())),
431            )
432            .finish();
433
434        let mut req = Request::put(&url);
435        // Set SSE headers.
436        req = self.insert_sse_headers(req);
437
438        if let Some(cache_control) = args.cache_control() {
439            req = req.header(constants::X_MS_BLOB_CACHE_CONTROL, cache_control);
440        }
441        if let Some(size) = size {
442            req = req.header(CONTENT_LENGTH, size)
443        }
444
445        if let Some(ty) = args.content_type() {
446            req = req.header(CONTENT_TYPE, ty)
447        }
448
449        let req = req
450            .extension(Operation::Write)
451            .body(body)
452            .map_err(new_request_build_error)?;
453
454        Ok(req)
455    }
456
457    pub async fn azblob_put_block(
458        &self,
459        path: &str,
460        block_id: Uuid,
461        size: Option<u64>,
462        args: &OpWrite,
463        body: Buffer,
464    ) -> Result<Response<Buffer>> {
465        let mut req = self.azblob_put_block_request(path, block_id, size, args, body)?;
466
467        self.sign(&mut req).await?;
468        self.send(req).await
469    }
470
471    fn azblob_complete_put_block_list_request(
472        &self,
473        path: &str,
474        block_ids: Vec<Uuid>,
475        args: &OpWrite,
476    ) -> Result<Request<Buffer>> {
477        let url = format!("{}?comp=blocklist", &self.build_path_url(path));
478
479        let req = Request::put(&url);
480
481        // Set SSE headers.
482        let mut req = self.insert_sse_headers(req);
483        if let Some(cache_control) = args.cache_control() {
484            req = req.header(constants::X_MS_BLOB_CACHE_CONTROL, cache_control);
485        }
486
487        let content = quick_xml::se::to_string(&PutBlockListRequest {
488            latest: block_ids
489                .into_iter()
490                .map(|block_id| {
491                    let encoded_block_id: String = BASE64_STANDARD.encode(block_id.as_bytes());
492                    encoded_block_id
493                })
494                .collect(),
495        })
496        .map_err(new_xml_serialize_error)?;
497
498        req = req.header(CONTENT_LENGTH, content.len());
499
500        let req = req
501            .extension(Operation::Write)
502            .body(Buffer::from(Bytes::from(content)))
503            .map_err(new_request_build_error)?;
504
505        Ok(req)
506    }
507
508    pub async fn azblob_complete_put_block_list(
509        &self,
510        path: &str,
511        block_ids: Vec<Uuid>,
512        args: &OpWrite,
513    ) -> Result<Response<Buffer>> {
514        let mut req = self.azblob_complete_put_block_list_request(path, block_ids, args)?;
515
516        self.sign(&mut req).await?;
517
518        self.send(req).await
519    }
520
521    pub fn azblob_head_blob_request(&self, path: &str, args: &OpStat) -> Result<Request<Buffer>> {
522        let mut req = Request::head(self.build_path_url(path));
523
524        // Set SSE headers.
525        req = self.insert_sse_headers(req);
526
527        if let Some(if_none_match) = args.if_none_match() {
528            req = req.header(IF_NONE_MATCH, if_none_match);
529        }
530
531        if let Some(if_match) = args.if_match() {
532            req = req.header(IF_MATCH, if_match);
533        }
534
535        let req = req
536            .extension(Operation::Stat)
537            .body(Buffer::new())
538            .map_err(new_request_build_error)?;
539
540        Ok(req)
541    }
542
543    pub async fn azblob_get_blob_properties(
544        &self,
545        path: &str,
546        args: &OpStat,
547    ) -> Result<Response<Buffer>> {
548        let mut req = self.azblob_head_blob_request(path, args)?;
549
550        self.sign(&mut req).await?;
551        self.send(req).await
552    }
553
554    fn azblob_delete_blob_request(&self, path: &str) -> Result<Request<Buffer>> {
555        Request::delete(self.build_path_url(path))
556            .header(CONTENT_LENGTH, 0)
557            .extension(Operation::Delete)
558            .body(Buffer::new())
559            .map_err(new_request_build_error)
560    }
561
562    pub async fn azblob_delete_blob(&self, path: &str) -> Result<Response<Buffer>> {
563        let mut req = self.azblob_delete_blob_request(path)?;
564
565        self.sign(&mut req).await?;
566        self.send(req).await
567    }
568
569    pub async fn azblob_copy_blob(
570        &self,
571        from: &str,
572        to: &str,
573        args: OpCopy,
574    ) -> Result<Response<Buffer>> {
575        let source = self.build_path_url(from);
576        let target = self.build_path_url(to);
577
578        let mut req = Request::put(&target)
579            .header(constants::X_MS_COPY_SOURCE, source)
580            .header(CONTENT_LENGTH, 0);
581
582        // Add if_not_exists condition using If-None-Match header
583        if args.if_not_exists() {
584            req = req.header(IF_NONE_MATCH, "*");
585        }
586
587        let mut req = req
588            .extension(Operation::Copy)
589            .body(Buffer::new())
590            .map_err(new_request_build_error)?;
591
592        self.sign(&mut req).await?;
593        self.send(req).await
594    }
595
596    pub async fn azblob_list_blobs(
597        &self,
598        path: &str,
599        next_marker: &str,
600        delimiter: &str,
601        limit: Option<usize>,
602    ) -> Result<Response<Buffer>> {
603        let p = build_abs_path(&self.root, path);
604        let mut url = QueryPairsWriter::new(&format!("{}/{}", self.endpoint, self.container))
605            .push("restype", "container")
606            .push("comp", "list");
607
608        if !p.is_empty() {
609            url = url.push("prefix", &percent_encode_path(&p));
610        }
611        if let Some(limit) = limit {
612            url = url.push("maxresults", &limit.to_string());
613        }
614        if !delimiter.is_empty() {
615            url = url.push("delimiter", delimiter);
616        }
617        if !next_marker.is_empty() {
618            url = url.push("marker", next_marker);
619        }
620
621        let mut req = Request::get(url.finish())
622            .extension(Operation::List)
623            .body(Buffer::new())
624            .map_err(new_request_build_error)?;
625
626        self.sign(&mut req).await?;
627        self.send(req).await
628    }
629
630    pub async fn azblob_batch_delete(&self, paths: &[String]) -> Result<Response<Buffer>> {
631        let url = format!(
632            "{}/{}?restype=container&comp=batch",
633            self.endpoint, self.container
634        );
635
636        let mut multipart = Multipart::new();
637
638        for (idx, path) in paths.iter().enumerate() {
639            let mut req = self.azblob_delete_blob_request(path)?;
640            self.batch_sign(&mut req).await?;
641
642            multipart = multipart.part(
643                MixedPart::from_request(req).part_header("content-id".parse().unwrap(), idx.into()),
644            );
645        }
646
647        let req = Request::post(url);
648        let mut req = multipart.apply(req)?;
649
650        self.sign(&mut req).await?;
651        self.send(req).await
652    }
653}
654
655/// Request of PutBlockListRequest
656#[derive(Default, Debug, Serialize, Deserialize)]
657#[serde(default, rename = "BlockList", rename_all = "PascalCase")]
658pub struct PutBlockListRequest {
659    pub latest: Vec<String>,
660}
661
662#[derive(Default, Debug, Deserialize)]
663#[serde(default, rename_all = "PascalCase")]
664pub struct ListBlobsOutput {
665    pub blobs: Blobs,
666    #[serde(rename = "NextMarker")]
667    pub next_marker: Option<String>,
668}
669
670#[derive(Default, Debug, Deserialize)]
671#[serde(default, rename_all = "PascalCase")]
672pub struct Blobs {
673    pub blob: Vec<Blob>,
674    pub blob_prefix: Vec<BlobPrefix>,
675}
676
677#[derive(Default, Debug, Deserialize)]
678#[serde(default, rename_all = "PascalCase")]
679pub struct BlobPrefix {
680    pub name: String,
681}
682
683#[derive(Default, Debug, Deserialize)]
684#[serde(default, rename_all = "PascalCase")]
685pub struct Blob {
686    pub properties: Properties,
687    pub name: String,
688}
689
690#[derive(Default, Debug, Deserialize)]
691#[serde(default, rename_all = "PascalCase")]
692pub struct Properties {
693    #[serde(rename = "Content-Length")]
694    pub content_length: u64,
695    #[serde(rename = "Last-Modified")]
696    pub last_modified: String,
697    #[serde(rename = "Content-MD5")]
698    pub content_md5: String,
699    #[serde(rename = "Content-Type")]
700    pub content_type: String,
701    pub etag: String,
702}
703
704#[cfg(test)]
705mod tests {
706    use bytes::Buf;
707    use bytes::Bytes;
708    use quick_xml::de;
709
710    use super::*;
711
712    #[test]
713    fn test_parse_xml() {
714        let bs = bytes::Bytes::from(
715            r#"
716            <?xml version="1.0" encoding="utf-8"?>
717            <EnumerationResults ServiceEndpoint="https://test.blob.core.windows.net/" ContainerName="myazurebucket">
718                <Prefix>dir1/</Prefix>
719                <Delimiter>/</Delimiter>
720                <Blobs>
721                    <Blob>
722                        <Name>dir1/2f018bb5-466f-4af1-84fa-2b167374ee06</Name>
723                        <Properties>
724                            <Creation-Time>Sun, 20 Mar 2022 11:29:03 GMT</Creation-Time>
725                            <Last-Modified>Sun, 20 Mar 2022 11:29:03 GMT</Last-Modified>
726                            <Etag>0x8DA0A64D66790C3</Etag>
727                            <Content-Length>3485277</Content-Length>
728                            <Content-Type>application/octet-stream</Content-Type>
729                            <Content-Encoding />
730                            <Content-Language />
731                            <Content-CRC64 />
732                            <Content-MD5>llJ/+jOlx5GdA1sL7SdKuw==</Content-MD5>
733                            <Cache-Control />
734                            <Content-Disposition />
735                            <BlobType>BlockBlob</BlobType>
736                            <AccessTier>Hot</AccessTier>
737                            <AccessTierInferred>true</AccessTierInferred>
738                            <LeaseStatus>unlocked</LeaseStatus>
739                            <LeaseState>available</LeaseState>
740                            <ServerEncrypted>true</ServerEncrypted>
741                        </Properties>
742                        <OrMetadata />
743                    </Blob>
744                    <Blob>
745                        <Name>dir1/5b9432b2-79c0-48d8-90c2-7d3e153826ed</Name>
746                        <Properties>
747                            <Creation-Time>Tue, 29 Mar 2022 01:54:07 GMT</Creation-Time>
748                            <Last-Modified>Tue, 29 Mar 2022 01:54:07 GMT</Last-Modified>
749                            <Etag>0x8DA112702D88FE4</Etag>
750                            <Content-Length>2471869</Content-Length>
751                            <Content-Type>application/octet-stream</Content-Type>
752                            <Content-Encoding />
753                            <Content-Language />
754                            <Content-CRC64 />
755                            <Content-MD5>xmgUltSnopLSJOukgCHFtg==</Content-MD5>
756                            <Cache-Control />
757                            <Content-Disposition />
758                            <BlobType>BlockBlob</BlobType>
759                            <AccessTier>Hot</AccessTier>
760                            <AccessTierInferred>true</AccessTierInferred>
761                            <LeaseStatus>unlocked</LeaseStatus>
762                            <LeaseState>available</LeaseState>
763                            <ServerEncrypted>true</ServerEncrypted>
764                        </Properties>
765                        <OrMetadata />
766                    </Blob>
767                    <Blob>
768                        <Name>dir1/b2d96f8b-d467-40d1-bb11-4632dddbf5b5</Name>
769                        <Properties>
770                            <Creation-Time>Sun, 20 Mar 2022 11:31:57 GMT</Creation-Time>
771                            <Last-Modified>Sun, 20 Mar 2022 11:31:57 GMT</Last-Modified>
772                            <Etag>0x8DA0A653DC82981</Etag>
773                            <Content-Length>1259677</Content-Length>
774                            <Content-Type>application/octet-stream</Content-Type>
775                            <Content-Encoding />
776                            <Content-Language />
777                            <Content-CRC64 />
778                            <Content-MD5>AxTiFXHwrXKaZC5b7ZRybw==</Content-MD5>
779                            <Cache-Control />
780                            <Content-Disposition />
781                            <BlobType>BlockBlob</BlobType>
782                            <AccessTier>Hot</AccessTier>
783                            <AccessTierInferred>true</AccessTierInferred>
784                            <LeaseStatus>unlocked</LeaseStatus>
785                            <LeaseState>available</LeaseState>
786                            <ServerEncrypted>true</ServerEncrypted>
787                        </Properties>
788                        <OrMetadata />
789                    </Blob>
790                    <BlobPrefix>
791                        <Name>dir1/dir2/</Name>
792                    </BlobPrefix>
793                    <BlobPrefix>
794                        <Name>dir1/dir21/</Name>
795                    </BlobPrefix>
796                </Blobs>
797                <NextMarker />
798            </EnumerationResults>"#,
799        );
800        let out: ListBlobsOutput = de::from_reader(bs.reader()).expect("must success");
801        println!("{out:?}");
802
803        assert_eq!(
804            out.blobs
805                .blob
806                .iter()
807                .map(|v| v.name.clone())
808                .collect::<Vec<String>>(),
809            [
810                "dir1/2f018bb5-466f-4af1-84fa-2b167374ee06",
811                "dir1/5b9432b2-79c0-48d8-90c2-7d3e153826ed",
812                "dir1/b2d96f8b-d467-40d1-bb11-4632dddbf5b5"
813            ]
814        );
815        assert_eq!(
816            out.blobs
817                .blob
818                .iter()
819                .map(|v| v.properties.content_length)
820                .collect::<Vec<u64>>(),
821            [3485277, 2471869, 1259677]
822        );
823        assert_eq!(
824            out.blobs
825                .blob
826                .iter()
827                .map(|v| v.properties.content_md5.clone())
828                .collect::<Vec<String>>(),
829            [
830                "llJ/+jOlx5GdA1sL7SdKuw==".to_string(),
831                "xmgUltSnopLSJOukgCHFtg==".to_string(),
832                "AxTiFXHwrXKaZC5b7ZRybw==".to_string()
833            ]
834        );
835        assert_eq!(
836            out.blobs
837                .blob
838                .iter()
839                .map(|v| v.properties.last_modified.clone())
840                .collect::<Vec<String>>(),
841            [
842                "Sun, 20 Mar 2022 11:29:03 GMT".to_string(),
843                "Tue, 29 Mar 2022 01:54:07 GMT".to_string(),
844                "Sun, 20 Mar 2022 11:31:57 GMT".to_string()
845            ]
846        );
847        assert_eq!(
848            out.blobs
849                .blob
850                .iter()
851                .map(|v| v.properties.etag.clone())
852                .collect::<Vec<String>>(),
853            [
854                "0x8DA0A64D66790C3".to_string(),
855                "0x8DA112702D88FE4".to_string(),
856                "0x8DA0A653DC82981".to_string()
857            ]
858        );
859        assert_eq!(
860            out.blobs
861                .blob_prefix
862                .iter()
863                .map(|v| v.name.clone())
864                .collect::<Vec<String>>(),
865            ["dir1/dir2/", "dir1/dir21/"]
866        );
867    }
868
869    /// This case is copied from real environment for testing
870    /// quick-xml overlapped-lists features. By default, quick-xml
871    /// can't deserialize content with overlapped-lists.
872    ///
873    /// For example, this case list blobs in this way:
874    ///
875    /// ```xml
876    /// <Blobs>
877    ///     <Blob>xxx</Blob>
878    ///     <BlobPrefix>yyy</BlobPrefix>
879    ///     <Blob>zzz</Blob>
880    /// </Blobs>
881    /// ```
882    ///
883    /// If `overlapped-lists` feature not enabled, we will get error `duplicate field Blob`.
884    #[test]
885    fn test_parse_overlapped_lists() {
886        let bs = "<?xml version=\"1.0\" encoding=\"utf-8\"?><EnumerationResults ServiceEndpoint=\"https://test.blob.core.windows.net/\" ContainerName=\"test\"><Prefix>9f7075e1-84d0-45ca-8196-ab9b71a8ef97/x/</Prefix><Delimiter>/</Delimiter><Blobs><Blob><Name>9f7075e1-84d0-45ca-8196-ab9b71a8ef97/x/</Name><Properties><Creation-Time>Thu, 01 Sep 2022 07:26:49 GMT</Creation-Time><Last-Modified>Thu, 01 Sep 2022 07:26:49 GMT</Last-Modified><Etag>0x8DA8BEB55D0EA35</Etag><Content-Length>0</Content-Length><Content-Type>application/octet-stream</Content-Type><Content-Encoding /><Content-Language /><Content-CRC64 /><Content-MD5>1B2M2Y8AsgTpgAmY7PhCfg==</Content-MD5><Cache-Control /><Content-Disposition /><BlobType>BlockBlob</BlobType><AccessTier>Hot</AccessTier><AccessTierInferred>true</AccessTierInferred><LeaseStatus>unlocked</LeaseStatus><LeaseState>available</LeaseState><ServerEncrypted>true</ServerEncrypted></Properties><OrMetadata /></Blob><BlobPrefix><Name>9f7075e1-84d0-45ca-8196-ab9b71a8ef97/x/x/</Name></BlobPrefix><Blob><Name>9f7075e1-84d0-45ca-8196-ab9b71a8ef97/x/y</Name><Properties><Creation-Time>Thu, 01 Sep 2022 07:26:50 GMT</Creation-Time><Last-Modified>Thu, 01 Sep 2022 07:26:50 GMT</Last-Modified><Etag>0x8DA8BEB55D99C08</Etag><Content-Length>0</Content-Length><Content-Type>application/octet-stream</Content-Type><Content-Encoding /><Content-Language /><Content-CRC64 /><Content-MD5>1B2M2Y8AsgTpgAmY7PhCfg==</Content-MD5><Cache-Control /><Content-Disposition /><BlobType>BlockBlob</BlobType><AccessTier>Hot</AccessTier><AccessTierInferred>true</AccessTierInferred><LeaseStatus>unlocked</LeaseStatus><LeaseState>available</LeaseState><ServerEncrypted>true</ServerEncrypted></Properties><OrMetadata /></Blob></Blobs><NextMarker /></EnumerationResults>";
887
888        de::from_reader(Bytes::from(bs).reader()).expect("must success")
889    }
890
891    /// This example is from https://learn.microsoft.com/en-us/rest/api/storageservices/put-block-list?tabs=microsoft-entra-id
892    #[test]
893    fn test_serialize_put_block_list_request() {
894        let req = PutBlockListRequest {
895            latest: vec!["1".to_string(), "2".to_string(), "3".to_string()],
896        };
897
898        let actual = quick_xml::se::to_string(&req).expect("must succeed");
899
900        pretty_assertions::assert_eq!(
901            actual,
902            r#"
903            <BlockList>
904               <Latest>1</Latest>
905               <Latest>2</Latest>
906               <Latest>3</Latest>
907            </BlockList>"#
908                // Cleanup space and new line
909                .replace([' ', '\n'], "")
910                // Escape `"` by hand to address <https://github.com/tafia/quick-xml/issues/362>
911                .replace('"', "&quot;")
912        );
913
914        let bs = "<?xml version=\"1.0\" encoding=\"utf-8\"?>
915            <BlockList>
916               <Latest>1</Latest>
917               <Latest>2</Latest>
918               <Latest>3</Latest>
919            </BlockList>";
920
921        let out: PutBlockListRequest =
922            de::from_reader(Bytes::from(bs).reader()).expect("must success");
923        assert_eq!(
924            out.latest,
925            vec!["1".to_string(), "2".to_string(), "3".to_string()]
926        );
927    }
928}
opendal_core/services/azblob/core.rs

opendal_core/services/azblob/
core.rs
