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

opendal/services/oss/
core.rs
