opendal_core/services/oss/
core.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::fmt::Debug;
19use std::sync::Arc;
20
21use bytes::Bytes;
22use constants::X_OSS_META_PREFIX;
23use http::HeaderMap;
24use http::HeaderName;
25use http::HeaderValue;
26use http::Request;
27use http::Response;
28use http::header::CACHE_CONTROL;
29use http::header::CONTENT_DISPOSITION;
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::header::RANGE;
37use reqsign::AliyunCredential;
38use reqsign::AliyunLoader;
39use reqsign::AliyunOssSigner;
40use serde::Deserialize;
41use serde::Serialize;
42
43use crate::raw::*;
44use crate::services::oss::core::constants::X_OSS_FORBID_OVERWRITE;
45use crate::*;
46
47pub mod constants {
48    pub const X_OSS_SERVER_SIDE_ENCRYPTION: &str = "x-oss-server-side-encryption";
49
50    pub const X_OSS_SERVER_SIDE_ENCRYPTION_KEY_ID: &str = "x-oss-server-side-encryption-key-id";
51
52    pub const X_OSS_FORBID_OVERWRITE: &str = "x-oss-forbid-overwrite";
53
54    pub const X_OSS_VERSION_ID: &str = "x-oss-version-id";
55
56    pub const RESPONSE_CONTENT_DISPOSITION: &str = "response-content-disposition";
57
58    pub const OSS_QUERY_VERSION_ID: &str = "versionId";
59
60    pub const X_OSS_META_PREFIX: &str = "x-oss-meta-";
61}
62
63pub struct OssCore {
64    pub info: Arc<AccessorInfo>,
65
66    pub root: String,
67    pub bucket: String,
68    /// buffered host string
69    ///
70    /// format: <bucket-name>.<endpoint-domain-name>
71    pub host: String,
72    pub endpoint: String,
73    pub presign_endpoint: String,
74    pub allow_anonymous: bool,
75
76    pub server_side_encryption: Option<HeaderValue>,
77    pub server_side_encryption_key_id: Option<HeaderValue>,
78
79    pub loader: AliyunLoader,
80    pub signer: AliyunOssSigner,
81}
82
83impl Debug for OssCore {
84    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
85        f.debug_struct("OssCore")
86            .field("root", &self.root)
87            .field("bucket", &self.bucket)
88            .field("endpoint", &self.endpoint)
89            .field("host", &self.host)
90            .finish_non_exhaustive()
91    }
92}
93
94impl OssCore {
95    async fn load_credential(&self) -> Result<Option<AliyunCredential>> {
96        let cred = self
97            .loader
98            .load()
99            .await
100            .map_err(new_request_credential_error)?;
101
102        if let Some(cred) = cred {
103            Ok(Some(cred))
104        } else if self.allow_anonymous {
105            // If allow_anonymous has been set, we will not sign the request.
106            Ok(None)
107        } else {
108            // Mark this error as temporary since it could be caused by Aliyun STS.
109            Err(Error::new(
110                ErrorKind::PermissionDenied,
111                "no valid credential found, please check configuration or try again",
112            )
113            .set_temporary())
114        }
115    }
116
117    pub async fn sign<T>(&self, req: &mut Request<T>) -> Result<()> {
118        let cred = if let Some(cred) = self.load_credential().await? {
119            cred
120        } else {
121            return Ok(());
122        };
123
124        self.signer.sign(req, &cred).map_err(new_request_sign_error)
125    }
126
127    pub async fn sign_query<T>(&self, req: &mut Request<T>, duration: Duration) -> Result<()> {
128        let cred = if let Some(cred) = self.load_credential().await? {
129            cred
130        } else {
131            return Ok(());
132        };
133
134        self.signer
135            .sign_query(req, duration, &cred)
136            .map_err(new_request_sign_error)
137    }
138
139    #[inline]
140    pub async fn send(&self, req: Request<Buffer>) -> Result<Response<Buffer>> {
141        self.info.http_client().send(req).await
142    }
143
144    /// Set sse headers
145    /// # Note
146    /// According to the OSS documentation, only PutObject, CopyObject, and InitiateMultipartUpload may require to be set.
147    pub fn insert_sse_headers(&self, mut req: http::request::Builder) -> http::request::Builder {
148        if let Some(v) = &self.server_side_encryption {
149            let mut v = v.clone();
150            v.set_sensitive(true);
151
152            req = req.header(
153                HeaderName::from_static(constants::X_OSS_SERVER_SIDE_ENCRYPTION),
154                v,
155            )
156        }
157        if let Some(v) = &self.server_side_encryption_key_id {
158            let mut v = v.clone();
159            v.set_sensitive(true);
160
161            req = req.header(
162                HeaderName::from_static(constants::X_OSS_SERVER_SIDE_ENCRYPTION_KEY_ID),
163                v,
164            )
165        }
166        req
167    }
168
169    fn insert_metadata_headers(
170        &self,
171        mut req: http::request::Builder,
172        size: Option<u64>,
173        args: &OpWrite,
174    ) -> Result<http::request::Builder> {
175        req = req.header(CONTENT_LENGTH, size.unwrap_or_default());
176
177        if let Some(mime) = args.content_type() {
178            req = req.header(CONTENT_TYPE, mime);
179        }
180
181        if let Some(pos) = args.content_disposition() {
182            req = req.header(CONTENT_DISPOSITION, pos);
183        }
184
185        if let Some(cache_control) = args.cache_control() {
186            req = req.header(CACHE_CONTROL, cache_control);
187        }
188
189        // TODO: disable if not exists while version has been enabled.
190        //
191        // Specifies whether the object that is uploaded by calling the PutObject operation
192        // overwrites the existing object that has the same name. When versioning is enabled
193        // or suspended for the bucket to which you want to upload the object, the
194        // x-oss-forbid-overwrite header does not take effect. In this case, the object that
195        // is uploaded by calling the PutObject operation overwrites the existing object that
196        // has the same name.
197        //
198        // ref: https://www.alibabacloud.com/help/en/oss/developer-reference/putobject?spm=a2c63.p38356.0.0.39ef75e93o0Xtz
199        if args.if_not_exists() {
200            req = req.header(X_OSS_FORBID_OVERWRITE, "true");
201        }
202
203        if let Some(user_metadata) = args.user_metadata() {
204            for (key, value) in user_metadata {
205                // before insert user defined metadata header, add prefix to the header name
206                if !self.check_user_metadata_key(key) {
207                    return Err(Error::new(
208                        ErrorKind::Unsupported,
209                        "the format of the user metadata key is invalid, please refer the document",
210                    ));
211                }
212                req = req.header(format!("{X_OSS_META_PREFIX}{key}"), value)
213            }
214        }
215
216        Ok(req)
217    }
218
219    // According to https://help.aliyun.com/zh/oss/developer-reference/putobject
220    // there are some limits in user defined metadata key
221    fn check_user_metadata_key(&self, key: &str) -> bool {
222        key.chars().all(|c| c.is_ascii_alphanumeric() || c == '-')
223    }
224
225    /// parse_metadata will parse http headers(including standards http headers
226    /// and user defined metadata header) into Metadata.
227    ///
228    /// # Arguments
229    ///
230    /// * `user_metadata_prefix` is the prefix of user defined metadata key
231    ///
232    /// # Notes
233    ///
234    /// before return the user defined metadata, we'll strip the user_metadata_prefix from the key
235    pub fn parse_metadata(&self, path: &str, headers: &HeaderMap) -> Result<Metadata> {
236        let mut m = parse_into_metadata(path, headers)?;
237        let user_meta = parse_prefixed_headers(headers, X_OSS_META_PREFIX);
238        if !user_meta.is_empty() {
239            m = m.with_user_metadata(user_meta);
240        }
241
242        Ok(m)
243    }
244}
245
246impl OssCore {
247    #[allow(clippy::too_many_arguments)]
248    pub fn oss_put_object_request(
249        &self,
250        path: &str,
251        size: Option<u64>,
252        args: &OpWrite,
253        body: Buffer,
254        is_presign: bool,
255    ) -> Result<Request<Buffer>> {
256        let p = build_abs_path(&self.root, path);
257        let endpoint = self.get_endpoint(is_presign);
258        let url = format!("{}/{}", endpoint, percent_encode_path(&p));
259
260        let mut req = Request::put(&url);
261
262        req = self.insert_metadata_headers(req, size, args)?;
263
264        // set sse headers
265        req = self.insert_sse_headers(req);
266
267        let req = req.extension(Operation::Write);
268
269        let req = req.body(body).map_err(new_request_build_error)?;
270        Ok(req)
271    }
272
273    pub fn oss_append_object_request(
274        &self,
275        path: &str,
276        position: u64,
277        size: u64,
278        args: &OpWrite,
279        body: Buffer,
280    ) -> Result<Request<Buffer>> {
281        let p = build_abs_path(&self.root, path);
282        let endpoint = self.get_endpoint(false);
283        let url = format!(
284            "{}/{}?append&position={}",
285            endpoint,
286            percent_encode_path(&p),
287            position
288        );
289
290        let mut req = Request::post(&url);
291
292        req = self.insert_metadata_headers(req, Some(size), args)?;
293
294        // set sse headers
295        req = self.insert_sse_headers(req);
296
297        let req = req.extension(Operation::Write);
298
299        let req = req.body(body).map_err(new_request_build_error)?;
300        Ok(req)
301    }
302
303    pub fn oss_get_object_request(
304        &self,
305        path: &str,
306        is_presign: bool,
307        args: &OpRead,
308    ) -> Result<Request<Buffer>> {
309        let p = build_abs_path(&self.root, path);
310        let endpoint = self.get_endpoint(is_presign);
311        let range = args.range();
312        let mut url = format!("{}/{}", endpoint, percent_encode_path(&p));
313
314        // Add query arguments to the URL based on response overrides
315        let mut query_args = Vec::new();
316        if let Some(override_content_disposition) = args.override_content_disposition() {
317            query_args.push(format!(
318                "{}={}",
319                constants::RESPONSE_CONTENT_DISPOSITION,
320                percent_encode_path(override_content_disposition)
321            ))
322        }
323        if let Some(version) = args.version() {
324            query_args.push(format!(
325                "{}={}",
326                constants::OSS_QUERY_VERSION_ID,
327                percent_encode_path(version)
328            ))
329        }
330
331        if !query_args.is_empty() {
332            url.push_str(&format!("?{}", query_args.join("&")));
333        }
334
335        let mut req = Request::get(&url);
336
337        if !range.is_full() {
338            req = req.header(RANGE, range.to_header());
339            // Adding `x-oss-range-behavior` header to use standard behavior.
340            // ref: https://help.aliyun.com/document_detail/39571.html
341            req = req.header("x-oss-range-behavior", "standard");
342        }
343
344        if let Some(if_match) = args.if_match() {
345            req = req.header(IF_MATCH, if_match)
346        }
347        if let Some(if_none_match) = args.if_none_match() {
348            req = req.header(IF_NONE_MATCH, if_none_match);
349        }
350
351        if let Some(if_modified_since) = args.if_modified_since() {
352            req = req.header(IF_MODIFIED_SINCE, if_modified_since.format_http_date());
353        }
354
355        if let Some(if_unmodified_since) = args.if_unmodified_since() {
356            req = req.header(IF_UNMODIFIED_SINCE, if_unmodified_since.format_http_date());
357        }
358
359        let req = req.extension(Operation::Read);
360
361        let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
362
363        Ok(req)
364    }
365
366    fn oss_delete_object_request(&self, path: &str, args: &OpDelete) -> Result<Request<Buffer>> {
367        let p = build_abs_path(&self.root, path);
368        let endpoint = self.get_endpoint(false);
369        let mut url = format!("{}/{}", endpoint, percent_encode_path(&p));
370
371        let mut query_args = Vec::new();
372
373        if let Some(version) = args.version() {
374            query_args.push(format!(
375                "{}={}",
376                constants::OSS_QUERY_VERSION_ID,
377                percent_encode_path(version)
378            ))
379        }
380
381        if !query_args.is_empty() {
382            url.push_str(&format!("?{}", query_args.join("&")));
383        }
384
385        let req = Request::delete(&url);
386
387        let req = req.extension(Operation::Delete);
388
389        let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
390
391        Ok(req)
392    }
393
394    pub fn oss_head_object_request(
395        &self,
396        path: &str,
397        is_presign: bool,
398        args: &OpStat,
399    ) -> Result<Request<Buffer>> {
400        let p = build_abs_path(&self.root, path);
401        let endpoint = self.get_endpoint(is_presign);
402        let mut url = format!("{}/{}", endpoint, percent_encode_path(&p));
403
404        let mut query_args = Vec::new();
405
406        if let Some(version) = args.version() {
407            query_args.push(format!(
408                "{}={}",
409                constants::OSS_QUERY_VERSION_ID,
410                percent_encode_path(version)
411            ))
412        }
413
414        if !query_args.is_empty() {
415            url.push_str(&format!("?{}", query_args.join("&")));
416        }
417
418        let mut req = Request::head(&url);
419        if let Some(if_match) = args.if_match() {
420            req = req.header(IF_MATCH, if_match)
421        }
422        if let Some(if_none_match) = args.if_none_match() {
423            req = req.header(IF_NONE_MATCH, if_none_match);
424        }
425
426        let req = req.extension(Operation::Stat);
427
428        let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
429
430        Ok(req)
431    }
432
433    pub fn oss_list_object_request(
434        &self,
435        path: &str,
436        token: &str,
437        delimiter: &str,
438        limit: Option<usize>,
439        start_after: Option<String>,
440    ) -> Result<Request<Buffer>> {
441        let p = build_abs_path(&self.root, path);
442
443        let endpoint = self.get_endpoint(false);
444        let mut url = QueryPairsWriter::new(endpoint);
445        url = url.push("list-type", "2");
446        if !delimiter.is_empty() {
447            url = url.push("delimiter", delimiter);
448        }
449        // prefix
450        if !p.is_empty() {
451            url = url.push("prefix", &percent_encode_path(&p));
452        }
453
454        // max-key
455        if let Some(limit) = limit {
456            url = url.push("max-keys", &limit.to_string());
457        }
458
459        // continuation_token
460        if !token.is_empty() {
461            url = url.push("continuation-token", &percent_encode_path(token));
462        }
463
464        // start-after
465        if let Some(start_after) = start_after {
466            let start_after = build_abs_path(&self.root, &start_after);
467            url = url.push("start-after", &percent_encode_path(&start_after));
468        }
469
470        let req = Request::get(url.finish())
471            .extension(Operation::List)
472            .body(Buffer::new())
473            .map_err(new_request_build_error)?;
474        Ok(req)
475    }
476
477    pub async fn oss_get_object(&self, path: &str, args: &OpRead) -> Result<Response<HttpBody>> {
478        let mut req = self.oss_get_object_request(path, false, args)?;
479        self.sign(&mut req).await?;
480        self.info.http_client().fetch(req).await
481    }
482
483    pub async fn oss_head_object(&self, path: &str, args: &OpStat) -> Result<Response<Buffer>> {
484        let mut req = self.oss_head_object_request(path, false, args)?;
485
486        self.sign(&mut req).await?;
487        self.send(req).await
488    }
489
490    pub async fn oss_copy_object(&self, from: &str, to: &str) -> Result<Response<Buffer>> {
491        let source = build_abs_path(&self.root, from);
492        let target = build_abs_path(&self.root, to);
493
494        let url = format!(
495            "{}/{}",
496            self.get_endpoint(false),
497            percent_encode_path(&target)
498        );
499        let source = format!("/{}/{}", self.bucket, percent_encode_path(&source));
500
501        let mut req = Request::put(&url);
502
503        req = self.insert_sse_headers(req);
504
505        req = req.header("x-oss-copy-source", source);
506
507        let req = req.extension(Operation::Copy);
508
509        let mut req = req.body(Buffer::new()).map_err(new_request_build_error)?;
510
511        self.sign(&mut req).await?;
512        self.send(req).await
513    }
514
515    pub async fn oss_list_object(
516        &self,
517        path: &str,
518        token: &str,
519        delimiter: &str,
520        limit: Option<usize>,
521        start_after: Option<String>,
522    ) -> Result<Response<Buffer>> {
523        let mut req = self.oss_list_object_request(path, token, delimiter, limit, start_after)?;
524
525        self.sign(&mut req).await?;
526        self.send(req).await
527    }
528
529    pub async fn oss_list_object_versions(
530        &self,
531        prefix: &str,
532        delimiter: &str,
533        limit: Option<usize>,
534        key_marker: &str,
535        version_id_marker: &str,
536    ) -> Result<Response<Buffer>> {
537        let p = build_abs_path(&self.root, prefix);
538
539        let mut url = QueryPairsWriter::new(&self.endpoint);
540        url = url.push("versions", "");
541
542        if !p.is_empty() {
543            url = url.push("prefix", &percent_encode_path(p.as_str()));
544        }
545        if !delimiter.is_empty() {
546            url = url.push("delimiter", delimiter);
547        }
548
549        if let Some(limit) = limit {
550            url = url.push("max-keys", &limit.to_string());
551        }
552        if !key_marker.is_empty() {
553            url = url.push("key-marker", &percent_encode_path(key_marker));
554        }
555        if !version_id_marker.is_empty() {
556            url = url.push("version-id-marker", &percent_encode_path(version_id_marker));
557        }
558
559        let mut req = Request::get(url.finish())
560            .extension(Operation::List)
561            .body(Buffer::new())
562            .map_err(new_request_build_error)?;
563
564        self.sign(&mut req).await?;
565
566        self.send(req).await
567    }
568
569    pub async fn oss_delete_object(&self, path: &str, args: &OpDelete) -> Result<Response<Buffer>> {
570        let mut req = self.oss_delete_object_request(path, args)?;
571        self.sign(&mut req).await?;
572        self.send(req).await
573    }
574
575    pub async fn oss_delete_objects(
576        &self,
577        paths: Vec<(String, OpDelete)>,
578    ) -> Result<Response<Buffer>> {
579        let url = format!("{}/?delete", self.endpoint);
580
581        let req = Request::post(&url);
582
583        let content = quick_xml::se::to_string(&DeleteObjectsRequest {
584            object: paths
585                .into_iter()
586                .map(|(path, op)| DeleteObjectsRequestObject {
587                    key: build_abs_path(&self.root, &path),
588                    version_id: op.version().map(|v| v.to_owned()),
589                })
590                .collect(),
591        })
592        .map_err(new_xml_serialize_error)?;
593
594        // Make sure content length has been set to avoid post with chunked encoding.
595        let req = req.header(CONTENT_LENGTH, content.len());
596        // Set content-type to `application/xml` to avoid mixed with form post.
597        let req = req.header(CONTENT_TYPE, "application/xml");
598        // Set content-md5 as required by API.
599        let req = req.header("CONTENT-MD5", format_content_md5(content.as_bytes()));
600
601        let req = req.extension(Operation::Delete);
602
603        let mut req = req
604            .body(Buffer::from(Bytes::from(content)))
605            .map_err(new_request_build_error)?;
606
607        self.sign(&mut req).await?;
608
609        self.send(req).await
610    }
611
612    fn get_endpoint(&self, is_presign: bool) -> &str {
613        if is_presign {
614            &self.presign_endpoint
615        } else {
616            &self.endpoint
617        }
618    }
619
620    pub async fn oss_initiate_upload(
621        &self,
622        path: &str,
623        content_type: Option<&str>,
624        content_disposition: Option<&str>,
625        cache_control: Option<&str>,
626        is_presign: bool,
627    ) -> Result<Response<Buffer>> {
628        let path = build_abs_path(&self.root, path);
629        let endpoint = self.get_endpoint(is_presign);
630        let url = format!("{}/{}?uploads", endpoint, percent_encode_path(&path));
631        let mut req = Request::post(&url);
632        if let Some(mime) = content_type {
633            req = req.header(CONTENT_TYPE, mime);
634        }
635        if let Some(disposition) = content_disposition {
636            req = req.header(CONTENT_DISPOSITION, disposition);
637        }
638        if let Some(cache_control) = cache_control {
639            req = req.header(CACHE_CONTROL, cache_control);
640        }
641        req = self.insert_sse_headers(req);
642
643        let req = req.extension(Operation::Write);
644
645        let mut req = req.body(Buffer::new()).map_err(new_request_build_error)?;
646        self.sign(&mut req).await?;
647        self.send(req).await
648    }
649
650    /// Creates a request to upload a part
651    pub async fn oss_upload_part_request(
652        &self,
653        path: &str,
654        upload_id: &str,
655        part_number: usize,
656        is_presign: bool,
657        size: u64,
658        body: Buffer,
659    ) -> Result<Response<Buffer>> {
660        let p = build_abs_path(&self.root, path);
661        let endpoint = self.get_endpoint(is_presign);
662
663        let url = format!(
664            "{}/{}?partNumber={}&uploadId={}",
665            endpoint,
666            percent_encode_path(&p),
667            part_number,
668            percent_encode_path(upload_id)
669        );
670
671        let mut req = Request::put(&url);
672        req = req.header(CONTENT_LENGTH, size);
673
674        let req = req.extension(Operation::Write);
675
676        let mut req = req.body(body).map_err(new_request_build_error)?;
677        self.sign(&mut req).await?;
678        self.send(req).await
679    }
680
681    pub async fn oss_complete_multipart_upload_request(
682        &self,
683        path: &str,
684        upload_id: &str,
685        is_presign: bool,
686        parts: Vec<MultipartUploadPart>,
687    ) -> Result<Response<Buffer>> {
688        let p = build_abs_path(&self.root, path);
689        let endpoint = self.get_endpoint(is_presign);
690        let url = format!(
691            "{}/{}?uploadId={}",
692            endpoint,
693            percent_encode_path(&p),
694            percent_encode_path(upload_id)
695        );
696
697        let req = Request::post(&url);
698
699        let content = quick_xml::se::to_string(&CompleteMultipartUploadRequest {
700            part: parts.to_vec(),
701        })
702        .map_err(new_xml_serialize_error)?;
703        // Make sure content length has been set to avoid post with chunked encoding.
704        let req = req.header(CONTENT_LENGTH, content.len());
705        // Set content-type to `application/xml` to avoid mixed with form post.
706        let req = req.header(CONTENT_TYPE, "application/xml");
707
708        let req = req.extension(Operation::Write);
709
710        let mut req = req
711            .body(Buffer::from(Bytes::from(content)))
712            .map_err(new_request_build_error)?;
713
714        self.sign(&mut req).await?;
715        self.send(req).await
716    }
717
718    /// Abort an ongoing multipart upload.
719    /// reference docs https://www.alibabacloud.com/help/zh/oss/developer-reference/abortmultipartupload
720    pub async fn oss_abort_multipart_upload(
721        &self,
722        path: &str,
723        upload_id: &str,
724    ) -> Result<Response<Buffer>> {
725        let p = build_abs_path(&self.root, path);
726
727        let url = format!(
728            "{}/{}?uploadId={}",
729            self.endpoint,
730            percent_encode_path(&p),
731            percent_encode_path(upload_id)
732        );
733
734        let mut req = Request::delete(&url)
735            .extension(Operation::Write)
736            .body(Buffer::new())
737            .map_err(new_request_build_error)?;
738        self.sign(&mut req).await?;
739        self.send(req).await
740    }
741}
742
743/// Request of DeleteObjects.
744#[derive(Default, Debug, Serialize)]
745#[serde(default, rename = "Delete", rename_all = "PascalCase")]
746pub struct DeleteObjectsRequest {
747    pub object: Vec<DeleteObjectsRequestObject>,
748}
749
750#[derive(Default, Debug, Serialize)]
751#[serde(rename_all = "PascalCase")]
752pub struct DeleteObjectsRequestObject {
753    pub key: String,
754    #[serde(skip_serializing_if = "Option::is_none")]
755    pub version_id: Option<String>,
756}
757
758/// Result of DeleteObjects.
759#[derive(Default, Debug, Deserialize)]
760#[serde(default, rename = "DeleteResult", rename_all = "PascalCase")]
761pub struct DeleteObjectsResult {
762    pub deleted: Vec<DeleteObjectsResultDeleted>,
763}
764
765#[derive(Default, Debug, Deserialize)]
766#[serde(rename_all = "PascalCase")]
767pub struct DeleteObjectsResultDeleted {
768    pub key: String,
769    pub version_id: Option<String>,
770}
771
772#[derive(Default, Debug, Deserialize)]
773#[serde(rename_all = "PascalCase")]
774pub struct InitiateMultipartUploadResult {
775    #[cfg(test)]
776    pub bucket: String,
777    #[cfg(test)]
778    pub key: String,
779    pub upload_id: String,
780}
781
782#[derive(Clone, Default, Debug, Serialize)]
783#[serde(default, rename_all = "PascalCase")]
784pub struct MultipartUploadPart {
785    #[serde(rename = "PartNumber")]
786    pub part_number: usize,
787    #[serde(rename = "ETag")]
788    pub etag: String,
789}
790
791#[derive(Default, Debug, Serialize)]
792#[serde(default, rename = "CompleteMultipartUpload", rename_all = "PascalCase")]
793pub struct CompleteMultipartUploadRequest {
794    pub part: Vec<MultipartUploadPart>,
795}
796
797#[derive(Default, Debug, Deserialize)]
798#[serde(default, rename_all = "PascalCase")]
799pub struct ListObjectsOutput {
800    pub prefix: String,
801    pub max_keys: u64,
802    pub encoding_type: String,
803    pub is_truncated: bool,
804    pub common_prefixes: Vec<CommonPrefix>,
805    pub contents: Vec<ListObjectsOutputContent>,
806    pub key_count: u64,
807
808    pub next_continuation_token: Option<String>,
809}
810
811#[derive(Default, Debug, Deserialize, PartialEq, Eq)]
812#[serde(default, rename_all = "PascalCase")]
813pub struct ListObjectsOutputContent {
814    pub key: String,
815    pub last_modified: String,
816    #[serde(rename = "ETag")]
817    pub etag: String,
818    pub size: u64,
819}
820
821#[derive(Default, Debug, Deserialize)]
822#[serde(default, rename_all = "PascalCase")]
823pub struct CommonPrefix {
824    pub prefix: String,
825}
826
827#[derive(Default, Debug, Eq, PartialEq, Deserialize)]
828#[serde(rename_all = "PascalCase")]
829pub struct OutputCommonPrefix {
830    pub prefix: String,
831}
832
833/// Output of ListObjectVersions
834#[derive(Default, Debug, Deserialize)]
835#[serde(default, rename_all = "PascalCase")]
836pub struct ListObjectVersionsOutput {
837    pub is_truncated: Option<bool>,
838    pub next_key_marker: Option<String>,
839    pub next_version_id_marker: Option<String>,
840    pub common_prefixes: Vec<OutputCommonPrefix>,
841    pub version: Vec<ListObjectVersionsOutputVersion>,
842    pub delete_marker: Vec<ListObjectVersionsOutputDeleteMarker>,
843}
844
845#[derive(Default, Debug, Eq, PartialEq, Deserialize)]
846#[serde(rename_all = "PascalCase")]
847pub struct ListObjectVersionsOutputVersion {
848    pub key: String,
849    pub version_id: String,
850    pub is_latest: bool,
851    pub size: u64,
852    pub last_modified: String,
853    #[serde(rename = "ETag")]
854    pub etag: Option<String>,
855}
856
857#[derive(Default, Debug, Eq, PartialEq, Deserialize)]
858#[serde(rename_all = "PascalCase")]
859pub struct ListObjectVersionsOutputDeleteMarker {
860    pub key: String,
861    pub version_id: String,
862    pub is_latest: bool,
863    pub last_modified: String,
864}
865
866#[cfg(test)]
867mod tests {
868    use bytes::Buf;
869    use bytes::Bytes;
870
871    use super::*;
872
873    /// This example is from https://www.alibabacloud.com/help/zh/object-storage-service/latest/deletemultipleobjects
874    #[test]
875    fn test_serialize_delete_objects_request() {
876        let req = DeleteObjectsRequest {
877            object: vec![
878                DeleteObjectsRequestObject {
879                    key: "multipart.data".to_string(),
880                    version_id: None,
881                },
882                DeleteObjectsRequestObject {
883                    key: "test.jpg".to_string(),
884                    version_id: None,
885                },
886                DeleteObjectsRequestObject {
887                    key: "demo.jpg".to_string(),
888                    version_id: None,
889                },
890            ],
891        };
892
893        let actual = quick_xml::se::to_string(&req).expect("must succeed");
894
895        pretty_assertions::assert_eq!(
896            actual,
897            r#"<Delete>
898  <Object>
899    <Key>multipart.data</Key>
900  </Object>
901  <Object>
902    <Key>test.jpg</Key>
903  </Object>
904  <Object>
905    <Key>demo.jpg</Key>
906  </Object>
907</Delete>"#
908                // Cleanup space and new line
909                .replace([' ', '\n'], "")
910        )
911    }
912
913    /// This example is from https://www.alibabacloud.com/help/zh/object-storage-service/latest/deletemultipleobjects
914    #[test]
915    fn test_deserialize_delete_objects_result() {
916        let bs = Bytes::from(
917            r#"<?xml version="1.0" encoding="UTF-8"?>
918<DeleteResult xmlns="http://doc.oss-cn-hangzhou.aliyuncs.com">
919    <Deleted>
920       <Key>multipart.data</Key>
921    </Deleted>
922    <Deleted>
923       <Key>test.jpg</Key>
924    </Deleted>
925    <Deleted>
926       <Key>demo.jpg</Key>
927    </Deleted>
928</DeleteResult>"#,
929        );
930
931        let out: DeleteObjectsResult =
932            quick_xml::de::from_reader(bs.reader()).expect("must success");
933
934        assert_eq!(out.deleted.len(), 3);
935        assert_eq!(out.deleted[0].key, "multipart.data");
936        assert_eq!(out.deleted[1].key, "test.jpg");
937        assert_eq!(out.deleted[2].key, "demo.jpg");
938    }
939
940    #[test]
941    fn test_deserialize_initiate_multipart_upload_response() {
942        let bs = Bytes::from(
943            r#"<?xml version="1.0" encoding="UTF-8"?>
944<InitiateMultipartUploadResult xmlns="http://doc.oss-cn-hangzhou.aliyuncs.com">
945    <Bucket>oss-example</Bucket>
946    <Key>multipart.data</Key>
947    <UploadId>0004B9894A22E5B1888A1E29F823****</UploadId>
948</InitiateMultipartUploadResult>"#,
949        );
950        let out: InitiateMultipartUploadResult =
951            quick_xml::de::from_reader(bs.reader()).expect("must success");
952
953        assert_eq!("0004B9894A22E5B1888A1E29F823****", out.upload_id);
954        assert_eq!("multipart.data", out.key);
955        assert_eq!("oss-example", out.bucket);
956    }
957
958    #[test]
959    fn test_serialize_complete_multipart_upload_request() {
960        let req = CompleteMultipartUploadRequest {
961            part: vec![
962                MultipartUploadPart {
963                    part_number: 1,
964                    etag: "\"3349DC700140D7F86A0784842780****\"".to_string(),
965                },
966                MultipartUploadPart {
967                    part_number: 5,
968                    etag: "\"8EFDA8BE206636A695359836FE0A****\"".to_string(),
969                },
970                MultipartUploadPart {
971                    part_number: 8,
972                    etag: "\"8C315065167132444177411FDA14****\"".to_string(),
973                },
974            ],
975        };
976
977        // quick_xml::se::to_string()
978        let mut serialized = String::new();
979        let mut serializer = quick_xml::se::Serializer::new(&mut serialized);
980        serializer.indent(' ', 4);
981        req.serialize(serializer).unwrap();
982        pretty_assertions::assert_eq!(
983            serialized,
984            r#"<CompleteMultipartUpload>
985    <Part>
986        <PartNumber>1</PartNumber>
987        <ETag>"3349DC700140D7F86A0784842780****"</ETag>
988    </Part>
989    <Part>
990        <PartNumber>5</PartNumber>
991        <ETag>"8EFDA8BE206636A695359836FE0A****"</ETag>
992    </Part>
993    <Part>
994        <PartNumber>8</PartNumber>
995        <ETag>"8C315065167132444177411FDA14****"</ETag>
996    </Part>
997</CompleteMultipartUpload>"#
998        )
999    }
1000
1001    #[test]
1002    fn test_parse_list_output() {
1003        let bs = bytes::Bytes::from(
1004            r#"<?xml version="1.0" encoding="UTF-8"?>
1005<ListBucketResult xmlns="https://doc.oss-cn-hangzhou.aliyuncs.com">
1006    <Name>examplebucket</Name>
1007    <Prefix></Prefix>
1008    <StartAfter>b</StartAfter>
1009    <MaxKeys>3</MaxKeys>
1010    <EncodingType>url</EncodingType>
1011    <IsTruncated>true</IsTruncated>
1012    <NextContinuationToken>CgJiYw--</NextContinuationToken>
1013    <Contents>
1014        <Key>b/c</Key>
1015        <LastModified>2020-05-18T05:45:54.000Z</LastModified>
1016        <ETag>"35A27C2B9EAEEB6F48FD7FB5861D****"</ETag>
1017        <Size>25</Size>
1018        <StorageClass>STANDARD</StorageClass>
1019        <Owner>
1020            <ID>1686240967192623</ID>
1021            <DisplayName>1686240967192623</DisplayName>
1022        </Owner>
1023    </Contents>
1024    <Contents>
1025        <Key>ba</Key>
1026        <LastModified>2020-05-18T11:17:58.000Z</LastModified>
1027        <ETag>"35A27C2B9EAEEB6F48FD7FB5861D****"</ETag>
1028        <Size>25</Size>
1029        <StorageClass>STANDARD</StorageClass>
1030        <Owner>
1031            <ID>1686240967192623</ID>
1032            <DisplayName>1686240967192623</DisplayName>
1033        </Owner>
1034    </Contents>
1035    <Contents>
1036        <Key>bc</Key>
1037        <LastModified>2020-05-18T05:45:59.000Z</LastModified>
1038        <ETag>"35A27C2B9EAEEB6F48FD7FB5861D****"</ETag>
1039        <Size>25</Size>
1040        <StorageClass>STANDARD</StorageClass>
1041        <Owner>
1042            <ID>1686240967192623</ID>
1043            <DisplayName>1686240967192623</DisplayName>
1044        </Owner>
1045    </Contents>
1046    <KeyCount>3</KeyCount>
1047</ListBucketResult>"#,
1048        );
1049
1050        let out: ListObjectsOutput = quick_xml::de::from_reader(bs.reader()).expect("must_success");
1051
1052        assert!(out.is_truncated);
1053        assert_eq!(out.next_continuation_token, Some("CgJiYw--".to_string()));
1054        assert!(out.common_prefixes.is_empty());
1055
1056        assert_eq!(
1057            out.contents,
1058            vec![
1059                ListObjectsOutputContent {
1060                    key: "b/c".to_string(),
1061                    last_modified: "2020-05-18T05:45:54.000Z".to_string(),
1062                    etag: "\"35A27C2B9EAEEB6F48FD7FB5861D****\"".to_string(),
1063                    size: 25,
1064                },
1065                ListObjectsOutputContent {
1066                    key: "ba".to_string(),
1067                    last_modified: "2020-05-18T11:17:58.000Z".to_string(),
1068                    etag: "\"35A27C2B9EAEEB6F48FD7FB5861D****\"".to_string(),
1069                    size: 25,
1070                },
1071                ListObjectsOutputContent {
1072                    key: "bc".to_string(),
1073                    last_modified: "2020-05-18T05:45:59.000Z".to_string(),
1074                    etag: "\"35A27C2B9EAEEB6F48FD7FB5861D****\"".to_string(),
1075                    size: 25,
1076                }
1077            ]
1078        )
1079    }
1080}
opendal_core/services/oss/core.rs

opendal_core/services/oss/
core.rs
