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

opendal/services/azblob/
core.rs
