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

opendal/services/azblob/
core.rs
