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

opendal/services/azblob/
core.rs
