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

opendal/services/azblob/
core.rs
