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    type Composer = ();
141
142    fn info(&self) -> ServiceInfo {
143        self.core.info.clone()
144    }
145
146    fn capability(&self) -> Capability {
147        self.core.capability
148    }
149
150    async fn create_dir(
151        &self,
152        ctx: &OperationContext,
153        path: &str,
154        _: OpCreateDir,
155    ) -> Result<RpCreateDir> {
156        self.core.ensure_dir_exists(ctx, path).await?;
157
158        Ok(RpCreateDir::default())
159    }
160
161    async fn rename(
162        &self,
163        ctx: &OperationContext,
164        from: &str,
165        to: &str,
166        _args: OpRename,
167    ) -> Result<RpRename> {
168        self.core.ensure_dir_exists(ctx, to).await?;
169
170        let resp = self.core.move_object(ctx, from, to).await?;
171
172        let status = resp.status();
173
174        match status {
175            StatusCode::OK | StatusCode::CREATED => Ok(RpRename::default()),
176            _ => Err(parse_error(
177                ErrorContext::new(ServiceOperation("MoveResource")),
178                resp,
179            )),
180        }
181    }
182
183    fn copy(
184        &self,
185        ctx: &OperationContext,
186        from: &str,
187        to: &str,
188        args: OpCopy,
189    ) -> Result<Self::Copier> {
190        let backend = self.clone();
191        let core = self.core.clone();
192        let ctx = ctx.clone();
193        let from = from.to_string();
194        let to = to.to_string();
195        let source_content_length_hint = args.source_content_length_hint();
196
197        Ok(oio::OneShotCopier::new(async move {
198            let source_size = match source_content_length_hint {
199                Some(size) => size,
200                None => backend
201                    .stat(&ctx, &from, OpStat::default())
202                    .await?
203                    .into_metadata()
204                    .content_length(),
205            };
206
207            core.ensure_dir_exists(&ctx, &to).await?;
208
209            let resp = core.copy(&ctx, &from, &to).await?;
210
211            let status = resp.status();
212
213            match status {
214                StatusCode::OK | StatusCode::CREATED => {
215                    Ok(MetadataBuilder::file(source_size).build())
216                }
217                _ => Err(parse_error(
218                    ErrorContext::new(ServiceOperation("CopyResource")),
219                    resp,
220                )),
221            }
222        }))
223    }
224    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
225        let output: oio::StreamReader<YandexDiskReader> = {
226            Ok(oio::StreamReader::new(YandexDiskReader::new(
227                self.clone(),
228                ctx.clone(),
229                path,
230                args,
231            )))
232        }?;
233
234        Ok(output)
235    }
236
237    async fn stat(&self, ctx: &OperationContext, path: &str, _args: OpStat) -> Result<RpStat> {
238        let resp = self.core.metainformation(ctx, path, None, None).await?;
239
240        let status = resp.status();
241
242        match status {
243            StatusCode::OK => {
244                let bs = resp.into_body();
245
246                let mf: MetainformationResponse =
247                    serde_json::from_reader(bs.reader()).map_err(new_json_deserialize_error)?;
248
249                parse_info(mf).map(RpStat::new)
250            }
251            _ => Err(parse_error(
252                ErrorContext::new(ServiceOperation("GetMetainformation")),
253                resp,
254            )),
255        }
256    }
257
258    fn write(&self, ctx: &OperationContext, path: &str, _args: OpWrite) -> Result<Self::Writer> {
259        let output: YandexDiskWriters = {
260            let writer = YandexDiskWriter::new(self.core.clone(), ctx.clone(), path.to_string());
261
262            let w = oio::OneShotWriter::new(writer);
263
264            Ok(w)
265        }?;
266
267        Ok(output)
268    }
269
270    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
271        let output: oio::OneShotDeleter<YandexDiskDeleter> = {
272            Ok(oio::OneShotDeleter::new(YandexDiskDeleter::new(
273                self.core.clone(),
274                ctx.clone(),
275            )))
276        }?;
277
278        Ok(output)
279    }
280
281    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
282        let output: oio::PageLister<YandexDiskLister> = {
283            let l = YandexDiskLister::new(self.core.clone(), ctx.clone(), path, args.limit());
284            Ok(oio::PageLister::new(l))
285        }?;
286
287        Ok(output)
288    }
289
290    async fn presign(
291        &self,
292        _ctx: &OperationContext,
293        _path: &str,
294        _args: OpPresign,
295    ) -> Result<RpPresign> {
296        Err(Error::new(
297            ErrorKind::Unsupported,
298            "operation is not supported",
299        ))
300    }
301}
opendal_service_yandex_disk/backend.rs

opendal_service_yandex_disk/
backend.rs
