opendal_service_oss/
backend.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 http::StatusCode;
22use http::Uri;
23use log::debug;
24use reqsign_aliyun_oss::AssumeRoleCredentialProvider;
25use reqsign_aliyun_oss::AssumeRoleWithOidcCredentialProvider;
26use reqsign_aliyun_oss::EcsRamRoleCredentialProvider;
27use reqsign_aliyun_oss::EnvCredentialProvider;
28use reqsign_aliyun_oss::RequestSigner;
29use reqsign_aliyun_oss::StaticCredentialProvider;
30use reqsign_core::Context;
31use reqsign_core::Env as _;
32use reqsign_core::OsEnv;
33use reqsign_core::ProvideCredentialChain;
34use reqsign_core::Signer;
35use reqsign_core::StaticEnv;
36use reqsign_file_read_tokio::TokioFileRead;
37
38use super::OSS_SCHEME;
39use super::config::OssConfig;
40use super::core::parse_error;
41use super::core::*;
42use super::deleter::OssDeleter;
43use super::lister::OssLister;
44use super::lister::OssListers;
45use super::lister::OssObjectVersionsLister;
46use super::reader::*;
47use super::writer::OssWriter;
48use super::writer::OssWriters;
49use opendal_core::raw::*;
50use opendal_core::*;
51
52const DEFAULT_BATCH_MAX_OPERATIONS: usize = 1000;
53
54/// Aliyun Object Storage Service (OSS) support
55#[doc = include_str!("docs.md")]
56#[derive(Default)]
57pub struct OssBuilder {
58    pub(super) config: OssConfig,
59}
60
61impl Debug for OssBuilder {
62    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
63        f.debug_struct("OssBuilder")
64            .field("config", &self.config)
65            .finish_non_exhaustive()
66    }
67}
68
69impl OssBuilder {
70    /// Set root of this backend.
71    ///
72    /// All operations will happen under this root.
73    pub fn root(mut self, root: &str) -> Self {
74        self.config.root = if root.is_empty() {
75            None
76        } else {
77            Some(root.to_string())
78        };
79
80        self
81    }
82
83    /// Set bucket name of this backend.
84    pub fn bucket(mut self, bucket: &str) -> Self {
85        self.config.bucket = bucket.to_string();
86
87        self
88    }
89
90    /// Set endpoint of this backend.
91    pub fn endpoint(mut self, endpoint: &str) -> Self {
92        if !endpoint.is_empty() {
93            // Trim trailing `/` so that we can accept `http://127.0.0.1:9000/`
94            self.config.endpoint = Some(endpoint.trim_end_matches('/').to_string())
95        }
96
97        self
98    }
99
100    /// Set addressing style for the endpoint.
101    ///
102    /// Available values: `virtual`, `cname`, `path`.
103    ///
104    /// - `virtual`: Use virtual addressing style, i.e. `http://bucket.oss-<region>.aliyuncs.com/object`
105    /// - `cname`: Use cname addressing style, i.e. `http://mydomain.com/object` with mydomain.com bound to your bucket.
106    /// - `path`: Use path addressing style. i.e. `http://oss-<region>.aliyuncs.com/bucket/object`
107    ///
108    /// - If not set, default value is `virtual`.
109    pub fn addressing_style(mut self, addressing_style: &str) -> Self {
110        self.config.addressing_style = Some(addressing_style.to_string());
111
112        self
113    }
114
115    /// Deprecated: OSS versioning capability is enabled by default.
116    #[deprecated(
117        since = "0.57.0",
118        note = "OSS versioning capability is enabled by default and this option is no longer needed."
119    )]
120    pub fn enable_versioning(self, _enabled: bool) -> Self {
121        self
122    }
123
124    /// Set an endpoint for generating presigned urls.
125    ///
126    /// You can offer a public endpoint like <https://oss-cn-beijing.aliyuncs.com> to return a presinged url for
127    /// public accessors, along with an internal endpoint like <https://oss-cn-beijing-internal.aliyuncs.com>
128    /// to access objects in a faster path.
129    ///
130    /// - If presign_endpoint is set, we will use presign_endpoint on generating presigned urls.
131    /// - if not, we will use endpoint as default.
132    pub fn presign_endpoint(mut self, endpoint: &str) -> Self {
133        if !endpoint.is_empty() {
134            // Trim trailing `/` so that we can accept `http://127.0.0.1:9000/`
135            self.config.presign_endpoint = Some(endpoint.trim_end_matches('/').to_string())
136        }
137
138        self
139    }
140
141    /// Set addressing style for presign endpoint.
142    ///
143    /// Similar to setting addressing style for endpoint.
144    ///
145    /// - If both presign_endpoint and presign_addressing_style are not set, they are the same as endpoint's configurations.
146    ///
147    /// - If presign_endpoint is set, but presign_addressing_style is not set, default value is `virtual`.
148    pub fn presign_addressing_style(mut self, addressing_style: &str) -> Self {
149        self.config.presign_addressing_style = Some(addressing_style.to_string());
150
151        self
152    }
153
154    /// Set access_key_id of this backend.
155    ///
156    /// - If access_key_id is set, we will take user's input first.
157    /// - If not, we will try to load it from environment.
158    pub fn access_key_id(mut self, v: &str) -> Self {
159        if !v.is_empty() {
160            self.config.access_key_id = Some(v.to_string())
161        }
162
163        self
164    }
165
166    /// Set access_key_secret of this backend.
167    ///
168    /// - If access_key_secret is set, we will take user's input first.
169    /// - If not, we will try to load it from environment.
170    pub fn access_key_secret(mut self, v: &str) -> Self {
171        if !v.is_empty() {
172            self.config.access_key_secret = Some(v.to_string())
173        }
174
175        self
176    }
177
178    /// Set security_token for this backend.
179    ///
180    /// - If security_token is set, we will take user's input first.
181    /// - If not, we will try to load it from environment.
182    pub fn security_token(mut self, security_token: &str) -> Self {
183        if !security_token.is_empty() {
184            self.config.security_token = Some(security_token.to_string())
185        }
186
187        self
188    }
189
190    /// preprocess the endpoint option
191    fn parse_endpoint(
192        &self,
193        endpoint: &Option<String>,
194        bucket: &str,
195        addressing_style: AddressingStyle,
196    ) -> Result<(String, String)> {
197        let (endpoint, host) = match endpoint.clone() {
198            Some(ep) => {
199                let uri = ep.parse::<Uri>().map_err(|err| {
200                    Error::new(ErrorKind::ConfigInvalid, "endpoint is invalid")
201                        .with_context("service", OSS_SCHEME)
202                        .with_context("endpoint", &ep)
203                        .set_source(err)
204                })?;
205                let host = uri.host().ok_or_else(|| {
206                    Error::new(ErrorKind::ConfigInvalid, "endpoint host is empty")
207                        .with_context("service", OSS_SCHEME)
208                        .with_context("endpoint", &ep)
209                })?;
210                let full_host = match addressing_style {
211                    AddressingStyle::Virtual => {
212                        if let Some(port) = uri.port_u16() {
213                            format!("{bucket}.{host}:{port}")
214                        } else {
215                            format!("{bucket}.{host}")
216                        }
217                    }
218                    AddressingStyle::Cname | AddressingStyle::Path => {
219                        if let Some(port) = uri.port_u16() {
220                            format!("{host}:{port}")
221                        } else {
222                            host.to_string()
223                        }
224                    }
225                };
226                if let Some(port) = uri.port_u16() {
227                    format!("{bucket}.{host}:{port}")
228                } else {
229                    format!("{bucket}.{host}")
230                };
231                let endpoint = match uri.scheme_str() {
232                    Some(scheme_str) => match scheme_str {
233                        "http" | "https" => format!("{scheme_str}://{full_host}"),
234                        _ => {
235                            return Err(Error::new(
236                                ErrorKind::ConfigInvalid,
237                                "endpoint protocol is invalid",
238                            )
239                            .with_context("service", OSS_SCHEME));
240                        }
241                    },
242                    None => format!("https://{full_host}"),
243                };
244                let endpoint = match addressing_style {
245                    AddressingStyle::Path => format!("{}/{}", endpoint, bucket),
246                    AddressingStyle::Cname | AddressingStyle::Virtual => endpoint,
247                };
248                (endpoint, full_host)
249            }
250            None => {
251                return Err(Error::new(ErrorKind::ConfigInvalid, "endpoint is empty")
252                    .with_context("service", OSS_SCHEME));
253            }
254        };
255        Ok((endpoint, host))
256    }
257
258    /// Set server_side_encryption for this backend.
259    ///
260    /// Available values: `AES256`, `KMS`.
261    ///
262    /// Reference: <https://www.alibabacloud.com/help/en/object-storage-service/latest/server-side-encryption-5>
263    /// Brief explanation:
264    /// There are two server-side encryption methods available:
265    /// SSE-AES256:
266    ///     1. Configure the bucket encryption mode as OSS-managed and specify the encryption algorithm as AES256.
267    ///     2. Include the `x-oss-server-side-encryption` parameter in the request and set its value to AES256.
268    /// SSE-KMS:
269    ///     1. To use this service, you need to first enable KMS.
270    ///     2. Configure the bucket encryption mode as KMS, and specify the specific CMK ID for BYOK (Bring Your Own Key)
271    ///        or not specify the specific CMK ID for OSS-managed KMS key.
272    ///     3. Include the `x-oss-server-side-encryption` parameter in the request and set its value to KMS.
273    ///     4. If a specific CMK ID is specified, include the `x-oss-server-side-encryption-key-id` parameter in the request, and set its value to the specified CMK ID.
274    pub fn server_side_encryption(mut self, v: &str) -> Self {
275        if !v.is_empty() {
276            self.config.server_side_encryption = Some(v.to_string())
277        }
278        self
279    }
280
281    /// Set server_side_encryption_key_id for this backend.
282    ///
283    /// # Notes
284    ///
285    /// This option only takes effect when server_side_encryption equals to KMS.
286    pub fn server_side_encryption_key_id(mut self, v: &str) -> Self {
287        if !v.is_empty() {
288            self.config.server_side_encryption_key_id = Some(v.to_string())
289        }
290        self
291    }
292
293    /// Deprecated: OSS delete batch capability is enabled by default.
294    #[deprecated(
295        since = "0.57.0",
296        note = "OSS delete batch capability is enabled by default and this option is no longer needed."
297    )]
298    pub fn batch_max_operations(self, _delete_max_size: usize) -> Self {
299        self
300    }
301
302    /// Deprecated: OSS delete batch capability is enabled by default.
303    #[deprecated(
304        since = "0.57.0",
305        note = "OSS delete batch capability is enabled by default and this option is no longer needed."
306    )]
307    pub fn delete_max_size(self, _delete_max_size: usize) -> Self {
308        self
309    }
310
311    /// Skip signature will skip loading credentials and signing requests.
312    pub fn skip_signature(mut self) -> Self {
313        self.config.skip_signature = true;
314        self
315    }
316
317    /// Allow anonymous will allow opendal to send request without signing
318    /// when credential is not loaded.
319    #[deprecated(
320        since = "0.57.0",
321        note = "Please use `skip_signature` instead of `allow_anonymous`"
322    )]
323    pub fn allow_anonymous(self) -> Self {
324        self.skip_signature()
325    }
326
327    /// Set role_arn for this backend.
328    ///
329    /// If `role_arn` is set, we will use already known config as source
330    /// credential to assume role with `role_arn`.
331    pub fn role_arn(mut self, role_arn: &str) -> Self {
332        if !role_arn.is_empty() {
333            self.config.role_arn = Some(role_arn.to_string())
334        }
335
336        self
337    }
338
339    /// Set role_session_name for this backend.
340    pub fn role_session_name(mut self, role_session_name: &str) -> Self {
341        if !role_session_name.is_empty() {
342            self.config.role_session_name = Some(role_session_name.to_string())
343        }
344
345        self
346    }
347
348    /// Set oidc_provider_arn for this backend.
349    pub fn oidc_provider_arn(mut self, oidc_provider_arn: &str) -> Self {
350        if !oidc_provider_arn.is_empty() {
351            self.config.oidc_provider_arn = Some(oidc_provider_arn.to_string())
352        }
353
354        self
355    }
356
357    /// Set oidc_token_file for this backend.
358    pub fn oidc_token_file(mut self, oidc_token_file: &str) -> Self {
359        if !oidc_token_file.is_empty() {
360            self.config.oidc_token_file = Some(oidc_token_file.to_string())
361        }
362
363        self
364    }
365
366    /// Set sts_endpoint for this backend.
367    pub fn sts_endpoint(mut self, sts_endpoint: &str) -> Self {
368        if !sts_endpoint.is_empty() {
369            self.config.sts_endpoint = Some(sts_endpoint.to_string())
370        }
371
372        self
373    }
374
375    /// Set external_id for this backend.
376    pub fn external_id(mut self, v: &str) -> Self {
377        if !v.is_empty() {
378            self.config.external_id = Some(v.to_string())
379        }
380
381        self
382    }
383}
384
385enum AddressingStyle {
386    Path,
387    Cname,
388    Virtual,
389}
390
391impl TryFrom<&Option<String>> for AddressingStyle {
392    type Error = Error;
393
394    fn try_from(value: &Option<String>) -> Result<Self> {
395        match value.as_deref() {
396            None | Some("virtual") => Ok(AddressingStyle::Virtual),
397            Some("path") => Ok(AddressingStyle::Path),
398            Some("cname") => Ok(AddressingStyle::Cname),
399            Some(v) => Err(Error::new(
400                ErrorKind::ConfigInvalid,
401                "Invalid addressing style, available: `virtual`, `path`, `cname`",
402            )
403            .with_context("service", OSS_SCHEME)
404            .with_context("addressing_style", v)),
405        }
406    }
407}
408
409impl Builder for OssBuilder {
410    type Config = OssConfig;
411
412    fn build(self) -> Result<impl Service> {
413        debug!("backend build started: {:?}", self);
414
415        #[allow(deprecated)]
416        let skip_signature = self.config.skip_signature || self.config.allow_anonymous;
417
418        let root = normalize_root(&self.config.root.clone().unwrap_or_default());
419        debug!("backend use root {}", root);
420
421        // Handle endpoint, region and bucket name.
422        let bucket = match self.config.bucket.is_empty() {
423            false => Ok(&self.config.bucket),
424            true => Err(
425                Error::new(ErrorKind::ConfigInvalid, "The bucket is misconfigured")
426                    .with_context("service", OSS_SCHEME),
427            ),
428        }?;
429
430        // Retrieve endpoint and host by parsing the endpoint option and bucket. If presign_endpoint is not
431        // set, take endpoint as default presign_endpoint.
432        let (endpoint, host) = self.parse_endpoint(
433            &self.config.endpoint,
434            bucket,
435            (&self.config.addressing_style).try_into()?,
436        )?;
437        debug!("backend use bucket {}, endpoint: {}", bucket, endpoint);
438
439        let presign_endpoint = if self.config.presign_endpoint.is_some() {
440            self.parse_endpoint(
441                &self.config.presign_endpoint,
442                bucket,
443                (&self.config.presign_addressing_style).try_into()?,
444            )?
445            .0
446        } else {
447            endpoint.clone()
448        };
449        debug!("backend use presign_endpoint: {}", presign_endpoint);
450
451        let server_side_encryption = match &self.config.server_side_encryption {
452            None => None,
453            Some(v) => Some(
454                build_header_value(v)
455                    .map_err(|err| err.with_context("key", "server_side_encryption"))?,
456            ),
457        };
458
459        let server_side_encryption_key_id = match &self.config.server_side_encryption_key_id {
460            None => None,
461            Some(v) => Some(
462                build_header_value(v)
463                    .map_err(|err| err.with_context("key", "server_side_encryption_key_id"))?,
464            ),
465        };
466
467        // NOTE: `AssumeRoleWithOidcCredentialProvider` still reads `role_arn`, `oidc_provider_arn`
468        // and `oidc_token_file` from `Context` environment variables at runtime. Until reqsign
469        // exposes typed builder APIs for all of them, we overlay config values into a `StaticEnv`
470        // snapshot here.
471        let os_env = OsEnv;
472        let mut envs = os_env.vars();
473
474        if let Some(v) = &self.config.role_arn {
475            envs.insert("ALIBABA_CLOUD_ROLE_ARN".to_string(), v.clone());
476        }
477        if let Some(v) = &self.config.oidc_provider_arn {
478            envs.insert("ALIBABA_CLOUD_OIDC_PROVIDER_ARN".to_string(), v.clone());
479        }
480        if let Some(v) = &self.config.oidc_token_file {
481            envs.insert("ALIBABA_CLOUD_OIDC_TOKEN_FILE".to_string(), v.clone());
482        }
483
484        let mut assume_role = AssumeRoleWithOidcCredentialProvider::new();
485
486        if let Some(sts_endpoint) = &self.config.sts_endpoint {
487            if sts_endpoint.starts_with("http://") || sts_endpoint.starts_with("https://") {
488                assume_role = assume_role.with_sts_endpoint(sts_endpoint.clone());
489            } else {
490                envs.insert(
491                    "ALIBABA_CLOUD_STS_ENDPOINT".to_string(),
492                    sts_endpoint.clone(),
493                );
494            }
495        }
496
497        if let Some(role_session_name) = &self.config.role_session_name {
498            assume_role = assume_role.with_role_session_name(role_session_name.clone());
499        }
500
501        let ctx = Context::new()
502            .with_file_read(TokioFileRead)
503            .with_env(StaticEnv {
504                home_dir: os_env.home_dir(),
505                envs,
506            });
507
508        let static_provider = if let (Some(ak), Some(sk)) =
509            (&self.config.access_key_id, &self.config.access_key_secret)
510        {
511            Some(if let Some(token) = self.config.security_token.as_deref() {
512                StaticCredentialProvider::new(ak, sk).with_security_token(token)
513            } else {
514                StaticCredentialProvider::new(ak, sk)
515            })
516        } else {
517            None
518        };
519
520        let mut provider = ProvideCredentialChain::new();
521        if let Some(static_provider) = static_provider {
522            provider = provider.push(static_provider);
523        }
524        let mut provider = provider
525            .push(EnvCredentialProvider::new())
526            .push(EcsRamRoleCredentialProvider::new())
527            .push(assume_role);
528
529        if let Some(role_arn) = &self.config.role_arn {
530            let mut assume_role_with_ak = AssumeRoleCredentialProvider::new()
531                .with_base_provider(provider)
532                .with_role_arn(role_arn.clone());
533
534            if let Some(role_session_name) = &self.config.role_session_name {
535                assume_role_with_ak =
536                    assume_role_with_ak.with_role_session_name(role_session_name.clone());
537            }
538            if let Some(external_id) = &self.config.external_id {
539                assume_role_with_ak = assume_role_with_ak.with_external_id(external_id.clone());
540            }
541            if let Some(sts_endpoint) = &self.config.sts_endpoint {
542                assume_role_with_ak = assume_role_with_ak.with_sts_endpoint(sts_endpoint.clone());
543            }
544
545            provider = ProvideCredentialChain::new().push(assume_role_with_ak);
546        }
547
548        let sign_ctx = ctx;
549        let request_signer = RequestSigner::new(bucket);
550        let signer = Signer::new(sign_ctx.clone(), provider, request_signer);
551
552        let info = ServiceInfo::new(OSS_SCHEME, &root, bucket);
553        let capability = Capability {
554            stat: true,
555            stat_with_if_match: true,
556            stat_with_if_none_match: true,
557            stat_with_version: true,
558
559            read: true,
560            read_with_suffix: true,
561
562            read_with_if_match: true,
563            read_with_if_none_match: true,
564            read_with_version: true,
565            read_with_if_modified_since: true,
566            read_with_if_unmodified_since: true,
567
568            write: true,
569            write_can_empty: true,
570            write_can_append: true,
571            write_can_multi: true,
572            write_with_cache_control: true,
573            write_with_content_type: true,
574            write_with_content_disposition: true,
575            write_with_content_encoding: true,
576            write_with_if_not_exists: true,
577
578            // The min multipart size of OSS is 100 KiB.
579            //
580            // ref: <https://www.alibabacloud.com/help/en/oss/user-guide/multipart-upload-12>
581            write_multi_min_size: Some(100 * 1024),
582            // The max multipart size of OSS is 5 GiB.
583            //
584            // ref: <https://www.alibabacloud.com/help/en/oss/user-guide/multipart-upload-12>
585            write_multi_max_size: if cfg!(target_pointer_width = "64") {
586                Some(5 * 1024 * 1024 * 1024)
587            } else {
588                Some(usize::MAX)
589            },
590            write_with_user_metadata: true,
591
592            delete: true,
593            delete_with_version: true,
594            delete_max_size: Some(DEFAULT_BATCH_MAX_OPERATIONS),
595
596            copy: true,
597
598            list: true,
599            list_with_limit: true,
600            list_with_start_after: true,
601            list_with_recursive: true,
602            list_with_versions: true,
603            list_with_deleted: true,
604
605            presign: true,
606            presign_stat: true,
607            presign_read: true,
608            presign_write: true,
609
610            shared: true,
611
612            ..Default::default()
613        };
614
615        Ok(OssBackend {
616            core: Arc::new(OssCore {
617                info,
618                capability,
619                root,
620                bucket: bucket.to_owned(),
621                endpoint,
622                host,
623                presign_endpoint,
624                skip_signature,
625                signer,
626                sign_ctx,
627                server_side_encryption,
628                server_side_encryption_key_id,
629            }),
630        })
631    }
632}
633
634#[derive(Debug, Clone)]
635/// Aliyun Object Storage Service backend
636pub struct OssBackend {
637    pub(crate) core: Arc<OssCore>,
638}
639
640impl Service for OssBackend {
641    type Reader = oio::StreamReader<OssReader>;
642    type Writer = OssWriters;
643    type Lister = OssListers;
644    type Deleter = oio::BatchDeleter<OssDeleter>;
645    type Copier = oio::OneShotCopier;
646    type Composer = ();
647
648    fn info(&self) -> ServiceInfo {
649        self.core.info.clone()
650    }
651
652    fn capability(&self) -> Capability {
653        self.core.capability
654    }
655
656    async fn create_dir(
657        &self,
658        _ctx: &OperationContext,
659        _path: &str,
660        _args: OpCreateDir,
661    ) -> Result<RpCreateDir> {
662        Err(Error::new(
663            ErrorKind::Unsupported,
664            "operation is not supported",
665        ))
666    }
667
668    async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
669        let resp = self.core.oss_head_object(ctx, path, &args).await?;
670
671        let status = resp.status();
672
673        match status {
674            StatusCode::OK => {
675                let headers = resp.headers();
676                let mut meta = self
677                    .core
678                    .parse_metadata(path, resp.headers())?
679                    .into_builder();
680
681                if let Some(v) = parse_header_to_str(headers, constants::X_OSS_VERSION_ID)? {
682                    meta.version(v);
683                }
684
685                Ok(RpStat::new(meta.build()))
686            }
687            _ => Err(parse_error(
688                ErrorContext::new(ServiceOperation("HeadObject"))
689                    .with_caller_condition(args.is_conditional()),
690                resp,
691            )),
692        }
693    }
694    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
695        let output: oio::StreamReader<OssReader> = {
696            Ok(oio::StreamReader::new(OssReader::new(
697                self.clone(),
698                ctx.clone(),
699                path,
700                args,
701            )))
702        }?;
703
704        Ok(output)
705    }
706
707    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
708        let output: OssWriters = {
709            let writer = OssWriter::new(self.core.clone(), ctx.clone(), path, args.clone());
710
711            let w = if args.append() {
712                OssWriters::Two(oio::AppendWriter::new(writer))
713            } else {
714                OssWriters::One(oio::MultipartWriter::new(
715                    ctx.executor().clone(),
716                    writer,
717                    args.concurrent(),
718                ))
719            };
720
721            Ok(w)
722        }?;
723
724        Ok(output)
725    }
726
727    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
728        let output: oio::BatchDeleter<OssDeleter> = {
729            Ok(oio::BatchDeleter::new(
730                OssDeleter::new(self.core.clone(), ctx.clone()),
731                self.core.capability.delete_max_size,
732            ))
733        }?;
734
735        Ok(output)
736    }
737
738    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
739        let output: OssListers = {
740            let l = if args.versions() || args.deleted() {
741                TwoWays::Two(oio::PageLister::new(OssObjectVersionsLister::new(
742                    self.core.clone(),
743                    ctx.clone(),
744                    path,
745                    args,
746                )))
747            } else {
748                TwoWays::One(oio::PageLister::new(OssLister::new(
749                    self.core.clone(),
750                    ctx.clone(),
751                    path,
752                    args.recursive(),
753                    args.limit(),
754                    args.start_after(),
755                )))
756            };
757
758            Ok(l)
759        }?;
760
761        Ok(output)
762    }
763
764    fn copy(
765        &self,
766        ctx: &OperationContext,
767        from: &str,
768        to: &str,
769        args: OpCopy,
770    ) -> Result<Self::Copier> {
771        let core = self.core.clone();
772        let ctx = ctx.clone();
773        let from = from.to_string();
774        let to = to.to_string();
775        Ok(oio::OneShotCopier::new(async move {
776            let source_size = match args.source_content_length_hint() {
777                Some(size) => size,
778                None => {
779                    let stat_args: OpStat = options::StatOptions {
780                        version: args.source_version().map(str::to_owned),
781                        ..Default::default()
782                    }
783                    .into();
784                    let resp = core.oss_head_object(&ctx, &from, &stat_args).await?;
785                    match resp.status() {
786                        StatusCode::OK => {
787                            parse_into_metadata(&from, resp.headers())?.content_length()
788                        }
789                        _ => {
790                            return Err(parse_error(
791                                ErrorContext::new(ServiceOperation("HeadObject")),
792                                resp,
793                            ));
794                        }
795                    }
796                }
797            };
798            let resp = core.oss_copy_object(&ctx, &from, &to).await?;
799            let status = resp.status();
800
801            match status {
802                StatusCode::OK => Ok(MetadataBuilder::file(source_size).build()),
803                _ => Err(parse_error(
804                    ErrorContext::new(ServiceOperation("CopyObject")),
805                    resp,
806                )),
807            }
808        }))
809    }
810
811    async fn rename(
812        &self,
813        _ctx: &OperationContext,
814        _from: &str,
815        _to: &str,
816        _args: OpRename,
817    ) -> Result<RpRename> {
818        Err(Error::new(
819            ErrorKind::Unsupported,
820            "operation is not supported",
821        ))
822    }
823
824    async fn presign(
825        &self,
826        ctx: &OperationContext,
827        path: &str,
828        args: OpPresign,
829    ) -> Result<RpPresign> {
830        // We will not send this request out, just for signing.
831        let req = match args.operation() {
832            PresignOperation::Stat(v) => self.core.oss_head_object_request(path, true, v),
833            PresignOperation::Read(range, v) => {
834                self.core.oss_get_object_request(path, true, *range, v)
835            }
836            PresignOperation::Write(v) => {
837                self.core
838                    .oss_put_object_request(path, None, v, Buffer::new(), true)
839            }
840            PresignOperation::Delete(_) => Err(Error::new(
841                ErrorKind::Unsupported,
842                "operation is not supported",
843            )),
844            _ => Err(Error::new(
845                ErrorKind::Unsupported,
846                "operation is not supported",
847            )),
848        };
849        let req = req?;
850        let req = self.core.sign_query(ctx, req, args.expire()).await?;
851
852        // We don't need this request anymore, consume it directly.
853        let (parts, _) = req.into_parts();
854
855        Ok(RpPresign::new(PresignedRequest::new(
856            parts.method,
857            parts.uri,
858            parts.headers,
859        )))
860    }
861}
opendal_service_oss/backend.rs

opendal_service_oss/
backend.rs
