opendal_service_yandex_disk/
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 bytes::Buf;
22use http::StatusCode;
23use log::debug;
24
25use super::YANDEX_DISK_SCHEME;
26use super::config::YandexDiskConfig;
27use super::core::parse_error;
28use super::core::*;
29use super::deleter::YandexDiskDeleter;
30use super::lister::YandexDiskLister;
31use super::reader::*;
32use super::writer::YandexDiskWriter;
33use super::writer::YandexDiskWriters;
34use opendal_core::raw::*;
35use opendal_core::*;
36
37/// [YandexDisk](https://360.yandex.com/disk/) services support.
38#[doc = include_str!("docs.md")]
39#[derive(Default)]
40pub struct YandexDiskBuilder {
41    pub(super) config: YandexDiskConfig,
42}
43
44impl Debug for YandexDiskBuilder {
45    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
46        f.debug_struct("YandexDiskBuilder")
47            .field("config", &self.config)
48            .finish_non_exhaustive()
49    }
50}
51
52impl YandexDiskBuilder {
53    /// Set root of this backend.
54    ///
55    /// All operations will happen under this root.
56    pub fn root(mut self, root: &str) -> Self {
57        self.config.root = if root.is_empty() {
58            None
59        } else {
60            Some(root.to_string())
61        };
62
63        self
64    }
65
66    /// yandex disk oauth access_token.
67    /// The valid token will looks like `y0_XXXXXXqihqIWAADLWwAAAAD3IXXXXXX0gtVeSPeIKM0oITMGhXXXXXX`.
68    /// We can fetch the debug token from <https://yandex.com/dev/disk/poligon>.
69    /// To use it in production, please register an app at <https://oauth.yandex.com> instead.
70    pub fn access_token(mut self, access_token: &str) -> Self {
71        self.config.access_token = access_token.to_string();
72
73        self
74    }
75}
76
77impl Builder for YandexDiskBuilder {
78    type Config = YandexDiskConfig;
79
80    /// Builds the backend and returns the result of YandexDiskBackend.
81    fn build(self) -> Result<impl Service> {
82        debug!("backend build started: {:?}", self);
83
84        let root = normalize_root(&self.config.root.clone().unwrap_or_default());
85        debug!("backend use root {}", root);
86
87        // Handle oauth access_token.
88        if self.config.access_token.is_empty() {
89            return Err(
90                Error::new(ErrorKind::ConfigInvalid, "access_token is empty")
91                    .with_operation("Builder::build")
92                    .with_context("service", YANDEX_DISK_SCHEME),
93            );
94        }
95
96        Ok(YandexDiskBackend {
97            core: Arc::new(YandexDiskCore {
98                info: ServiceInfo::new(YANDEX_DISK_SCHEME, &root, ""),
99                capability: Capability {
100                    stat: true,
101
102                    create_dir: true,
103
104                    read: true,
105                    read_with_suffix: true,
106
107                    write: true,
108                    write_can_empty: true,
109
110                    delete: true,
111                    rename: true,
112                    copy: true,
113
114                    list: true,
115                    list_with_limit: true,
116
117                    shared: true,
118
119                    ..Default::default()
120                },
121                root,
122                access_token: self.config.access_token.clone(),
123            }),
124        })
125    }
126}
127
128/// Backend for YandexDisk services.
129#[derive(Debug, Clone)]
130pub struct YandexDiskBackend {
131    pub(crate) core: Arc<YandexDiskCore>,
132}
133
134impl Service for YandexDiskBackend {
135    type Reader = oio::StreamReader<YandexDiskReader>;
136    type Writer = YandexDiskWriters;
137    type Lister = oio::PageLister<YandexDiskLister>;
138    type Deleter = oio::OneShotDeleter<YandexDiskDeleter>;
139    type Copier = oio::OneShotCopier;
140
141    fn info(&self) -> ServiceInfo {
142        self.core.info.clone()
143    }
144
145    fn capability(&self) -> Capability {
146        self.core.capability
147    }
148
149    async fn create_dir(
150        &self,
151        ctx: &OperationContext,
152        path: &str,
153        _: OpCreateDir,
154    ) -> Result<RpCreateDir> {
155        self.core.ensure_dir_exists(ctx, path).await?;
156
157        Ok(RpCreateDir::default())
158    }
159
160    async fn rename(
161        &self,
162        ctx: &OperationContext,
163        from: &str,
164        to: &str,
165        _args: OpRename,
166    ) -> Result<RpRename> {
167        self.core.ensure_dir_exists(ctx, to).await?;
168
169        let resp = self.core.move_object(ctx, from, to).await?;
170
171        let status = resp.status();
172
173        match status {
174            StatusCode::OK | StatusCode::CREATED => Ok(RpRename::default()),
175            _ => Err(parse_error(resp)),
176        }
177    }
178
179    fn copy(
180        &self,
181        ctx: &OperationContext,
182        from: &str,
183        to: &str,
184        _args: OpCopy,
185        _opts: OpCopier,
186    ) -> Result<Self::Copier> {
187        let core = self.core.clone();
188        let ctx = ctx.clone();
189        let from = from.to_string();
190        let to = to.to_string();
191
192        Ok(oio::OneShotCopier::new(async move {
193            core.ensure_dir_exists(&ctx, &to).await?;
194
195            let resp = core.copy(&ctx, &from, &to).await?;
196
197            let status = resp.status();
198
199            match status {
200                StatusCode::OK | StatusCode::CREATED => Ok(Metadata::default()),
201                _ => Err(parse_error(resp)),
202            }
203        }))
204    }
205    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
206        let output: oio::StreamReader<YandexDiskReader> = {
207            Ok(oio::StreamReader::new(YandexDiskReader::new(
208                self.clone(),
209                ctx.clone(),
210                path,
211                args,
212            )))
213        }?;
214
215        Ok(output)
216    }
217
218    async fn stat(&self, ctx: &OperationContext, path: &str, _args: OpStat) -> Result<RpStat> {
219        let resp = self.core.metainformation(ctx, path, None, None).await?;
220
221        let status = resp.status();
222
223        match status {
224            StatusCode::OK => {
225                let bs = resp.into_body();
226
227                let mf: MetainformationResponse =
228                    serde_json::from_reader(bs.reader()).map_err(new_json_deserialize_error)?;
229
230                parse_info(mf).map(RpStat::new)
231            }
232            _ => Err(parse_error(resp)),
233        }
234    }
235
236    fn write(&self, ctx: &OperationContext, path: &str, _args: OpWrite) -> Result<Self::Writer> {
237        let output: YandexDiskWriters = {
238            let writer = YandexDiskWriter::new(self.core.clone(), ctx.clone(), path.to_string());
239
240            let w = oio::OneShotWriter::new(writer);
241
242            Ok(w)
243        }?;
244
245        Ok(output)
246    }
247
248    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
249        let output: oio::OneShotDeleter<YandexDiskDeleter> = {
250            Ok(oio::OneShotDeleter::new(YandexDiskDeleter::new(
251                self.core.clone(),
252                ctx.clone(),
253            )))
254        }?;
255
256        Ok(output)
257    }
258
259    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
260        let output: oio::PageLister<YandexDiskLister> = {
261            let l = YandexDiskLister::new(self.core.clone(), ctx.clone(), path, args.limit());
262            Ok(oio::PageLister::new(l))
263        }?;
264
265        Ok(output)
266    }
267
268    async fn presign(
269        &self,
270        _ctx: &OperationContext,
271        _path: &str,
272        _args: OpPresign,
273    ) -> Result<RpPresign> {
274        Err(Error::new(
275            ErrorKind::Unsupported,
276            "operation is not supported",
277        ))
278    }
279}
opendal_service_yandex_disk/backend.rs

opendal_service_yandex_disk/
backend.rs
