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

opendal/services/oss/
core.rs
