opendal/services/gcs/
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::collections::HashMap;
19use std::fmt::Debug;
20use std::fmt::Formatter;
21use std::fmt::Write;
22use std::sync::Arc;
23use std::sync::LazyLock;
24use std::time::Duration;
25
26use backon::ExponentialBuilder;
27use backon::Retryable;
28use bytes::Buf;
29use bytes::Bytes;
30use constants::*;
31use http::header::CACHE_CONTROL;
32use http::header::CONTENT_DISPOSITION;
33use http::header::CONTENT_ENCODING;
34use http::header::CONTENT_LENGTH;
35use http::header::CONTENT_TYPE;
36use http::header::HOST;
37use http::header::IF_MATCH;
38use http::header::IF_MODIFIED_SINCE;
39use http::header::IF_NONE_MATCH;
40use http::header::IF_UNMODIFIED_SINCE;
41use http::Request;
42use http::Response;
43use reqsign::GoogleCredential;
44use reqsign::GoogleCredentialLoader;
45use reqsign::GoogleSigner;
46use reqsign::GoogleToken;
47use reqsign::GoogleTokenLoader;
48use serde::Deserialize;
49use serde::Serialize;
50
51use super::uri::percent_encode_path;
52use crate::raw::*;
53use crate::*;
54
55pub mod constants {
56    pub const X_GOOG_ACL: &str = "x-goog-acl";
57    pub const X_GOOG_STORAGE_CLASS: &str = "x-goog-storage-class";
58    pub const X_GOOG_META_PREFIX: &str = "x-goog-meta-";
59}
60
61pub struct GcsCore {
62    pub info: Arc<AccessorInfo>,
63    pub endpoint: String,
64    pub bucket: String,
65    pub root: String,
66
67    pub signer: GoogleSigner,
68    pub token_loader: GoogleTokenLoader,
69    pub token: Option<String>,
70    pub scope: String,
71    pub credential_loader: GoogleCredentialLoader,
72
73    pub predefined_acl: Option<String>,
74    pub default_storage_class: Option<String>,
75
76    pub allow_anonymous: bool,
77}
78
79impl Debug for GcsCore {
80    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
81        let mut de = f.debug_struct("Backend");
82        de.field("endpoint", &self.endpoint)
83            .field("bucket", &self.bucket)
84            .field("root", &self.root)
85            .finish_non_exhaustive()
86    }
87}
88
89static BACKOFF: LazyLock<ExponentialBuilder> =
90    LazyLock::new(|| ExponentialBuilder::default().with_jitter());
91
92impl GcsCore {
93    async fn load_token(&self) -> Result<Option<GoogleToken>> {
94        if let Some(token) = &self.token {
95            return Ok(Some(GoogleToken::new(token, usize::MAX, &self.scope)));
96        }
97
98        let cred = { || self.token_loader.load() }
99            .retry(*BACKOFF)
100            .await
101            .map_err(new_request_credential_error)?;
102
103        if let Some(cred) = cred {
104            return Ok(Some(cred));
105        }
106
107        if self.allow_anonymous {
108            return Ok(None);
109        }
110
111        Err(Error::new(
112            ErrorKind::ConfigInvalid,
113            "no valid credential found",
114        ))
115    }
116
117    fn load_credential(&self) -> Result<Option<GoogleCredential>> {
118        let cred = self
119            .credential_loader
120            .load()
121            .map_err(new_request_credential_error)?;
122
123        if let Some(cred) = cred {
124            return Ok(Some(cred));
125        }
126
127        if self.allow_anonymous {
128            return Ok(None);
129        }
130
131        Err(Error::new(
132            ErrorKind::ConfigInvalid,
133            "no valid credential found",
134        ))
135    }
136
137    pub async fn sign<T>(&self, req: &mut Request<T>) -> Result<()> {
138        if let Some(cred) = self.load_token().await? {
139            self.signer
140                .sign(req, &cred)
141                .map_err(new_request_sign_error)?;
142        } else {
143            return Ok(());
144        }
145
146        // Always remove host header, let users' client to set it based on HTTP
147        // version.
148        //
149        // As discussed in <https://github.com/seanmonstar/reqwest/issues/1809>,
150        // google server could send RST_STREAM of PROTOCOL_ERROR if our request
151        // contains host header.
152        req.headers_mut().remove(HOST);
153
154        Ok(())
155    }
156
157    pub fn sign_query<T>(&self, req: &mut Request<T>, duration: Duration) -> Result<()> {
158        if let Some(cred) = self.load_credential()? {
159            self.signer
160                .sign_query(req, duration, &cred)
161                .map_err(new_request_sign_error)?;
162        } else {
163            return Ok(());
164        }
165
166        // Always remove host header, let users' client to set it based on HTTP
167        // version.
168        //
169        // As discussed in <https://github.com/seanmonstar/reqwest/issues/1809>,
170        // google server could send RST_STREAM of PROTOCOL_ERROR if our request
171        // contains host header.
172        req.headers_mut().remove(HOST);
173
174        Ok(())
175    }
176
177    #[inline]
178    pub async fn send(&self, req: Request<Buffer>) -> Result<Response<Buffer>> {
179        self.info.http_client().send(req).await
180    }
181}
182
183impl GcsCore {
184    pub fn gcs_get_object_request(
185        &self,
186        path: &str,
187        range: BytesRange,
188        args: &OpRead,
189    ) -> Result<Request<Buffer>> {
190        let p = build_abs_path(&self.root, path);
191
192        let url = format!(
193            "{}/storage/v1/b/{}/o/{}?alt=media",
194            self.endpoint,
195            self.bucket,
196            percent_encode_path(&p)
197        );
198
199        let mut req = Request::get(&url);
200
201        if let Some(if_match) = args.if_match() {
202            req = req.header(IF_MATCH, if_match);
203        }
204        if let Some(if_none_match) = args.if_none_match() {
205            req = req.header(IF_NONE_MATCH, if_none_match);
206        }
207        if !range.is_full() {
208            req = req.header(http::header::RANGE, range.to_header());
209        }
210
211        let req = req.extension(Operation::Read);
212
213        let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
214
215        Ok(req)
216    }
217
218    // It's for presign operation. Gcs only supports query sign over XML API.
219    pub fn gcs_get_object_xml_request(&self, path: &str, args: &OpRead) -> Result<Request<Buffer>> {
220        let p = build_abs_path(&self.root, path);
221
222        let url = format!("{}/{}/{}", self.endpoint, self.bucket, p);
223
224        let mut req = Request::get(&url);
225
226        if let Some(if_match) = args.if_match() {
227            req = req.header(IF_MATCH, if_match);
228        }
229        if let Some(if_none_match) = args.if_none_match() {
230            req = req.header(IF_NONE_MATCH, if_none_match);
231        }
232
233        if let Some(if_modified_since) = args.if_modified_since() {
234            req = req.header(
235                IF_MODIFIED_SINCE,
236                format_datetime_into_http_date(if_modified_since),
237            );
238        }
239
240        if let Some(if_unmodified_since) = args.if_unmodified_since() {
241            req = req.header(
242                IF_UNMODIFIED_SINCE,
243                format_datetime_into_http_date(if_unmodified_since),
244            );
245        }
246
247        let req = req.extension(Operation::Read);
248
249        let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
250
251        Ok(req)
252    }
253
254    pub async fn gcs_get_object(
255        &self,
256        path: &str,
257        range: BytesRange,
258        args: &OpRead,
259    ) -> Result<Response<HttpBody>> {
260        let mut req = self.gcs_get_object_request(path, range, args)?;
261
262        self.sign(&mut req).await?;
263        self.info.http_client().fetch(req).await
264    }
265
266    pub fn gcs_insert_object_request(
267        &self,
268        path: &str,
269        size: Option<u64>,
270        op: &OpWrite,
271        body: Buffer,
272    ) -> Result<Request<Buffer>> {
273        let p = build_abs_path(&self.root, path);
274
275        let request_metadata = InsertRequestMetadata {
276            storage_class: self.default_storage_class.as_deref(),
277            cache_control: op.cache_control(),
278            content_type: op.content_type(),
279            content_encoding: op.content_encoding(),
280            metadata: op.user_metadata(),
281        };
282
283        let mut url = format!(
284            "{}/upload/storage/v1/b/{}/o?uploadType={}&name={}",
285            self.endpoint,
286            self.bucket,
287            if request_metadata.is_empty() {
288                "media"
289            } else {
290                "multipart"
291            },
292            percent_encode_path(&p)
293        );
294
295        if let Some(acl) = &self.predefined_acl {
296            write!(&mut url, "&predefinedAcl={}", acl).unwrap();
297        }
298
299        // Makes the operation conditional on whether the object's current generation
300        // matches the given value. Setting to 0 makes the operation succeed only if
301        // there are no live versions of the object.
302        if op.if_not_exists() {
303            write!(&mut url, "&ifGenerationMatch=0").unwrap();
304        }
305
306        let mut req = Request::post(&url);
307
308        req = req.header(CONTENT_LENGTH, size.unwrap_or_default());
309
310        if request_metadata.is_empty() {
311            let req = req.extension(Operation::Write);
312            // If the metadata is empty, we do not set any `Content-Type` header,
313            // since if we had it in the `op.content_type()`, it would be already set in the
314            // `multipart` metadata body and this branch won't be executed.
315            let req = req.body(body).map_err(new_request_build_error)?;
316            Ok(req)
317        } else {
318            let mut multipart = Multipart::new();
319            let metadata_part = RelatedPart::new()
320                .header(
321                    CONTENT_TYPE,
322                    "application/json; charset=UTF-8".parse().unwrap(),
323                )
324                .content(
325                    serde_json::to_vec(&request_metadata)
326                        .expect("metadata serialization should succeed"),
327                );
328            multipart = multipart.part(metadata_part);
329
330            // Content-Type must be set, even if it is set in the metadata part
331            let content_type = op
332                .content_type()
333                .unwrap_or("application/octet-stream")
334                .parse()
335                .expect("Failed to parse content-type");
336            let media_part = RelatedPart::new()
337                .header(CONTENT_TYPE, content_type)
338                .content(body);
339            multipart = multipart.part(media_part);
340
341            let req = multipart.apply(Request::post(url).extension(Operation::Write))?;
342
343            Ok(req)
344        }
345    }
346
347    // It's for presign operation. Gcs only supports query sign over XML API.
348    pub fn gcs_insert_object_xml_request(
349        &self,
350        path: &str,
351        args: &OpWrite,
352        body: Buffer,
353    ) -> Result<Request<Buffer>> {
354        let p = build_abs_path(&self.root, path);
355
356        let url = format!("{}/{}/{}", self.endpoint, self.bucket, p);
357
358        let mut req = Request::put(&url);
359
360        if let Some(user_metadata) = args.user_metadata() {
361            for (key, value) in user_metadata {
362                req = req.header(format!("{X_GOOG_META_PREFIX}{key}"), value)
363            }
364        }
365
366        if let Some(content_type) = args.content_type() {
367            req = req.header(CONTENT_TYPE, content_type);
368        }
369
370        if let Some(content_encoding) = args.content_encoding() {
371            req = req.header(CONTENT_ENCODING, content_encoding);
372        }
373
374        if let Some(acl) = &self.predefined_acl {
375            if let Some(predefined_acl_in_xml_spec) = predefined_acl_to_xml_header(acl) {
376                req = req.header(X_GOOG_ACL, predefined_acl_in_xml_spec);
377            } else {
378                log::warn!("Unrecognized predefined_acl. Ignoring");
379            }
380        }
381
382        if let Some(storage_class) = &self.default_storage_class {
383            req = req.header(X_GOOG_STORAGE_CLASS, storage_class);
384        }
385
386        let req = req.extension(Operation::Write);
387
388        let req = req.body(body).map_err(new_request_build_error)?;
389
390        Ok(req)
391    }
392
393    pub fn gcs_head_object_request(&self, path: &str, args: &OpStat) -> Result<Request<Buffer>> {
394        let p = build_abs_path(&self.root, path);
395
396        let url = format!(
397            "{}/storage/v1/b/{}/o/{}",
398            self.endpoint,
399            self.bucket,
400            percent_encode_path(&p)
401        );
402
403        let mut req = Request::get(&url);
404
405        if let Some(if_none_match) = args.if_none_match() {
406            req = req.header(IF_NONE_MATCH, if_none_match);
407        }
408
409        if let Some(if_match) = args.if_match() {
410            req = req.header(IF_MATCH, if_match);
411        }
412
413        let req = req.extension(Operation::Stat);
414
415        let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
416
417        Ok(req)
418    }
419
420    // It's for presign operation. Gcs only supports query sign over XML API.
421    pub fn gcs_head_object_xml_request(
422        &self,
423        path: &str,
424        args: &OpStat,
425    ) -> Result<Request<Buffer>> {
426        let p = build_abs_path(&self.root, path);
427
428        let url = format!("{}/{}/{}", self.endpoint, self.bucket, p);
429
430        let mut req = Request::head(&url);
431
432        if let Some(if_none_match) = args.if_none_match() {
433            req = req.header(IF_NONE_MATCH, if_none_match);
434        }
435
436        if let Some(if_match) = args.if_match() {
437            req = req.header(IF_MATCH, if_match);
438        }
439
440        let req = req.extension(Operation::Stat);
441
442        let req = req.body(Buffer::new()).map_err(new_request_build_error)?;
443
444        Ok(req)
445    }
446
447    pub async fn gcs_get_object_metadata(
448        &self,
449        path: &str,
450        args: &OpStat,
451    ) -> Result<Response<Buffer>> {
452        let mut req = self.gcs_head_object_request(path, args)?;
453
454        self.sign(&mut req).await?;
455
456        self.send(req).await
457    }
458
459    pub async fn gcs_delete_object(&self, path: &str) -> Result<Response<Buffer>> {
460        let mut req = self.gcs_delete_object_request(path)?;
461
462        self.sign(&mut req).await?;
463        self.send(req).await
464    }
465
466    pub fn gcs_delete_object_request(&self, path: &str) -> Result<Request<Buffer>> {
467        let p = build_abs_path(&self.root, path);
468
469        let url = format!(
470            "{}/storage/v1/b/{}/o/{}",
471            self.endpoint,
472            self.bucket,
473            percent_encode_path(&p)
474        );
475
476        Request::delete(&url)
477            .body(Buffer::new())
478            .map_err(new_request_build_error)
479    }
480
481    pub async fn gcs_delete_objects(&self, paths: Vec<String>) -> Result<Response<Buffer>> {
482        let uri = format!("{}/batch/storage/v1", self.endpoint);
483
484        let mut multipart = Multipart::new();
485
486        for (idx, path) in paths.iter().enumerate() {
487            let req = self.gcs_delete_object_request(path)?;
488
489            multipart = multipart.part(
490                MixedPart::from_request(req).part_header("content-id".parse().unwrap(), idx.into()),
491            );
492        }
493
494        let req = Request::post(uri).extension(Operation::Delete);
495        let mut req = multipart.apply(req)?;
496
497        self.sign(&mut req).await?;
498        self.send(req).await
499    }
500
501    pub async fn gcs_copy_object(&self, from: &str, to: &str) -> Result<Response<Buffer>> {
502        let source = build_abs_path(&self.root, from);
503        let dest = build_abs_path(&self.root, to);
504
505        let req_uri = format!(
506            "{}/storage/v1/b/{}/o/{}/copyTo/b/{}/o/{}",
507            self.endpoint,
508            self.bucket,
509            percent_encode_path(&source),
510            self.bucket,
511            percent_encode_path(&dest)
512        );
513
514        let mut req = Request::post(req_uri)
515            .header(CONTENT_LENGTH, 0)
516            .extension(Operation::Copy)
517            .body(Buffer::new())
518            .map_err(new_request_build_error)?;
519
520        self.sign(&mut req).await?;
521        self.send(req).await
522    }
523
524    pub async fn gcs_list_objects(
525        &self,
526        path: &str,
527        page_token: &str,
528        delimiter: &str,
529        limit: Option<usize>,
530        start_after: Option<String>,
531    ) -> Result<Response<Buffer>> {
532        let p = build_abs_path(&self.root, path);
533
534        let url = format!("{}/storage/v1/b/{}/o", self.endpoint, self.bucket,);
535
536        let mut url = QueryPairsWriter::new(&url);
537        url = url.push("prefix", &percent_encode_path(&p));
538
539        if !delimiter.is_empty() {
540            url = url.push("delimiter", delimiter);
541        }
542        if let Some(limit) = limit {
543            url = url.push("maxResults", &limit.to_string());
544        }
545        if let Some(start_after) = start_after {
546            let start_after = build_abs_path(&self.root, &start_after);
547            url = url.push("startOffset", &percent_encode_path(&start_after));
548        }
549
550        if !page_token.is_empty() {
551            // NOTE:
552            //
553            // GCS uses pageToken in request and nextPageToken in response
554            //
555            // Don't know how will those tokens be like so this part are copied
556            // directly from AWS S3 service.
557            url = url.push("pageToken", &percent_encode_path(page_token));
558        }
559
560        let mut req = Request::get(url.finish())
561            .extension(Operation::List)
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 gcs_initiate_multipart_upload(
571        &self,
572        path: &str,
573        op: &OpWrite,
574    ) -> Result<Response<Buffer>> {
575        let p = build_abs_path(&self.root, path);
576
577        let url = format!("{}/{}/{}?uploads", self.endpoint, self.bucket, p);
578
579        let mut builder = Request::post(&url)
580            .header(CONTENT_LENGTH, 0)
581            .extension(Operation::Write);
582
583        if let Some(header_val) = op.content_disposition() {
584            builder = builder.header(CONTENT_DISPOSITION, header_val);
585        }
586
587        if let Some(header_val) = op.content_encoding() {
588            builder = builder.header(CONTENT_ENCODING, header_val);
589        }
590
591        if let Some(header_val) = op.content_type() {
592            builder = builder.header(CONTENT_TYPE, header_val);
593        }
594
595        if let Some(header_val) = op.cache_control() {
596            builder = builder.header(CACHE_CONTROL, header_val);
597        }
598
599        if let Some(metadata) = op.user_metadata() {
600            for (k, v) in metadata {
601                builder = builder.header(&format!("x-goog-meta-{k}"), v);
602            }
603        }
604
605        if let Some(acl) = self.predefined_acl.as_ref() {
606            if let Some(predefined_acl_in_xml_spec) = predefined_acl_to_xml_header(acl) {
607                builder = builder.header(X_GOOG_ACL, predefined_acl_in_xml_spec);
608            } else {
609                log::warn!("Unrecognized predefined_acl. Ignoring");
610            }
611        }
612
613        let mut req = builder
614            .body(Buffer::new())
615            .map_err(new_request_build_error)?;
616
617        self.sign(&mut req).await?;
618        self.send(req).await
619    }
620
621    pub async fn gcs_upload_part(
622        &self,
623        path: &str,
624        upload_id: &str,
625        part_number: usize,
626        size: u64,
627        body: Buffer,
628    ) -> Result<Response<Buffer>> {
629        let p = build_abs_path(&self.root, path);
630
631        let url = format!(
632            "{}/{}/{}?partNumber={}&uploadId={}",
633            self.endpoint,
634            self.bucket,
635            percent_encode_path(&p),
636            part_number,
637            percent_encode_path(upload_id)
638        );
639
640        let mut req = Request::put(&url);
641
642        req = req.header(CONTENT_LENGTH, size);
643
644        let req = req.extension(Operation::Write);
645
646        let mut req = req.body(body).map_err(new_request_build_error)?;
647
648        self.sign(&mut req).await?;
649        self.send(req).await
650    }
651
652    pub async fn gcs_complete_multipart_upload(
653        &self,
654        path: &str,
655        upload_id: &str,
656        parts: Vec<CompleteMultipartUploadRequestPart>,
657    ) -> Result<Response<Buffer>> {
658        let p = build_abs_path(&self.root, path);
659
660        let url = format!(
661            "{}/{}/{}?uploadId={}",
662            self.endpoint,
663            self.bucket,
664            percent_encode_path(&p),
665            percent_encode_path(upload_id)
666        );
667
668        let req = Request::post(&url);
669
670        let content = quick_xml::se::to_string(&CompleteMultipartUploadRequest { part: parts })
671            .map_err(new_xml_serialize_error)?;
672        // Make sure content length has been set to avoid post with chunked encoding.
673        let req = req.header(CONTENT_LENGTH, content.len());
674        // Set content-type to `application/xml` to avoid mixed with form post.
675        let req = req.header(CONTENT_TYPE, "application/xml");
676
677        let req = req.extension(Operation::Write);
678
679        let mut req = req
680            .body(Buffer::from(Bytes::from(content)))
681            .map_err(new_request_build_error)?;
682
683        self.sign(&mut req).await?;
684        self.send(req).await
685    }
686
687    pub async fn gcs_abort_multipart_upload(
688        &self,
689        path: &str,
690        upload_id: &str,
691    ) -> Result<Response<Buffer>> {
692        let p = build_abs_path(&self.root, path);
693
694        let url = format!(
695            "{}/{}/{}?uploadId={}",
696            self.endpoint,
697            self.bucket,
698            percent_encode_path(&p),
699            percent_encode_path(upload_id)
700        );
701
702        let mut req = Request::delete(&url)
703            .extension(Operation::Write)
704            .body(Buffer::new())
705            .map_err(new_request_build_error)?;
706        self.sign(&mut req).await?;
707        self.send(req).await
708    }
709
710    pub fn build_metadata_from_object_response(path: &str, data: Buffer) -> Result<Metadata> {
711        let meta: GetObjectJsonResponse =
712            serde_json::from_reader(data.reader()).map_err(new_json_deserialize_error)?;
713
714        let mut m = Metadata::new(EntryMode::from_path(path));
715
716        m.set_etag(&meta.etag);
717        m.set_content_md5(&meta.md5_hash);
718
719        let size = meta
720            .size
721            .parse::<u64>()
722            .map_err(|e| Error::new(ErrorKind::Unexpected, "parse u64").set_source(e))?;
723        m.set_content_length(size);
724        if !meta.content_type.is_empty() {
725            m.set_content_type(&meta.content_type);
726        }
727
728        if !meta.content_encoding.is_empty() {
729            m.set_content_encoding(&meta.content_encoding);
730        }
731
732        if !meta.cache_control.is_empty() {
733            m.set_cache_control(&meta.cache_control);
734        }
735
736        if !meta.content_disposition.is_empty() {
737            m.set_content_disposition(&meta.content_disposition);
738        }
739
740        if !meta.generation.is_empty() {
741            m.set_version(&meta.generation);
742        }
743
744        m.set_last_modified(parse_datetime_from_rfc3339(&meta.updated)?);
745
746        if !meta.metadata.is_empty() {
747            m = m.with_user_metadata(meta.metadata);
748        }
749
750        Ok(m)
751    }
752}
753
754// https://cloud.google.com/storage/docs/xml-api/reference-headers#xgoogacl
755fn predefined_acl_to_xml_header(predefined_acl: &str) -> Option<&'static str> {
756    match predefined_acl {
757        "projectPrivate" => Some("project-private"),
758        "private" => Some("private"),
759        "bucketOwnerRead" => Some("bucket-owner-read"),
760        "bucketOwnerFullControl" => Some("bucket-owner-full-control"),
761        "publicRead" => Some("public-read"),
762        "authenticatedRead" => Some("authenticated-read"),
763        _ => None,
764    }
765}
766
767#[derive(Debug, Serialize)]
768#[serde(default, rename_all = "camelCase")]
769pub struct InsertRequestMetadata<'a> {
770    #[serde(skip_serializing_if = "Option::is_none")]
771    content_type: Option<&'a str>,
772    #[serde(skip_serializing_if = "Option::is_none")]
773    content_encoding: Option<&'a str>,
774    #[serde(skip_serializing_if = "Option::is_none")]
775    storage_class: Option<&'a str>,
776    #[serde(skip_serializing_if = "Option::is_none")]
777    cache_control: Option<&'a str>,
778    #[serde(skip_serializing_if = "Option::is_none")]
779    metadata: Option<&'a HashMap<String, String>>,
780}
781
782impl InsertRequestMetadata<'_> {
783    pub fn is_empty(&self) -> bool {
784        self.content_type.is_none()
785            && self.content_encoding.is_none()
786            && self.storage_class.is_none()
787            && self.cache_control.is_none()
788            // We could also put content-encoding in the url parameters
789            && self.content_encoding.is_none()
790            && self.metadata.is_none()
791    }
792}
793/// Response JSON from GCS list objects API.
794///
795/// refer to https://cloud.google.com/storage/docs/json_api/v1/objects/list for details
796#[derive(Default, Debug, Deserialize)]
797#[serde(default, rename_all = "camelCase")]
798pub struct ListResponse {
799    /// The continuation token.
800    ///
801    /// If this is the last page of results, then no continuation token is returned.
802    pub next_page_token: Option<String>,
803    /// Object name prefixes for objects that matched the listing request
804    /// but were excluded from [items] because of a delimiter.
805    pub prefixes: Vec<String>,
806    /// The list of objects, ordered lexicographically by name.
807    pub items: Vec<ListResponseItem>,
808}
809
810#[derive(Default, Debug, Eq, PartialEq, Deserialize)]
811#[serde(default, rename_all = "camelCase")]
812pub struct ListResponseItem {
813    pub name: String,
814    pub size: String,
815    // metadata
816    pub etag: String,
817    pub md5_hash: String,
818    pub updated: String,
819    pub content_type: String,
820}
821
822/// Result of CreateMultipartUpload
823#[derive(Default, Debug, Deserialize)]
824#[serde(default, rename_all = "PascalCase")]
825pub struct InitiateMultipartUploadResult {
826    pub upload_id: String,
827}
828
829/// Request of CompleteMultipartUploadRequest
830#[derive(Default, Debug, Serialize)]
831#[serde(default, rename = "CompleteMultipartUpload", rename_all = "PascalCase")]
832pub struct CompleteMultipartUploadRequest {
833    pub part: Vec<CompleteMultipartUploadRequestPart>,
834}
835
836#[derive(Clone, Default, Debug, Serialize)]
837#[serde(default, rename_all = "PascalCase")]
838pub struct CompleteMultipartUploadRequestPart {
839    #[serde(rename = "PartNumber")]
840    pub part_number: usize,
841    #[serde(rename = "ETag")]
842    pub etag: String,
843}
844
845/// The raw json response returned by [`get`](https://cloud.google.com/storage/docs/json_api/v1/objects/get)
846#[derive(Debug, Default, Deserialize)]
847#[serde(default, rename_all = "camelCase")]
848struct GetObjectJsonResponse {
849    /// GCS will return size in string.
850    ///
851    /// For example: `"size": "56535"`
852    size: String,
853    /// etag is not quoted.
854    ///
855    /// For example: `"etag": "CKWasoTgyPkCEAE="`
856    etag: String,
857    /// RFC3339 styled datetime string.
858    ///
859    /// For example: `"updated": "2022-08-15T11:33:34.866Z"`
860    updated: String,
861    /// Content md5 hash
862    ///
863    /// For example: `"md5Hash": "fHcEH1vPwA6eTPqxuasXcg=="`
864    md5_hash: String,
865    /// Content type of this object.
866    ///
867    /// For example: `"contentType": "image/png",`
868    content_type: String,
869    /// Content encoding of this object
870    ///
871    /// For example: "contentEncoding": "br"
872    content_encoding: String,
873    /// Content disposition of this object
874    content_disposition: String,
875    /// Cache-Control directive for the object data.
876    cache_control: String,
877    /// Content generation of this object. Used for object versioning and soft delete.
878    generation: String,
879    /// Custom metadata of this object.
880    ///
881    /// For example: `"metadata" : { "my-key": "my-value" }`
882    metadata: HashMap<String, String>,
883}
884
885#[cfg(test)]
886mod tests {
887    use super::*;
888
889    #[test]
890    fn test_deserialize_get_object_json_response() {
891        let content = r#"{
892    "kind": "storage#object",
893    "id": "example/1.png/1660563214863653",
894    "selfLink": "https://www.googleapis.com/storage/v1/b/example/o/1.png",
895    "mediaLink": "https://content-storage.googleapis.com/download/storage/v1/b/example/o/1.png?generation=1660563214863653&alt=media",
896    "name": "1.png",
897    "bucket": "example",
898    "generation": "1660563214863653",
899    "metageneration": "1",
900    "contentType": "image/png",
901    "contentEncoding": "br",
902    "contentDisposition": "attachment",
903    "cacheControl": "public, max-age=3600",
904    "storageClass": "STANDARD",
905    "size": "56535",
906    "md5Hash": "fHcEH1vPwA6eTPqxuasXcg==",
907    "crc32c": "j/un9g==",
908    "etag": "CKWasoTgyPkCEAE=",
909    "timeCreated": "2022-08-15T11:33:34.866Z",
910    "updated": "2022-08-15T11:33:34.866Z",
911    "timeStorageClassUpdated": "2022-08-15T11:33:34.866Z",
912    "metadata" : {
913        "location" : "everywhere"
914  }
915}"#;
916
917        let meta = GcsCore::build_metadata_from_object_response("1.png", content.into())
918            .expect("parse metadata should not fail");
919
920        assert_eq!(meta.content_length(), 56535);
921        assert_eq!(
922            meta.last_modified(),
923            Some(
924                parse_datetime_from_rfc3339("2022-08-15T11:33:34.866Z")
925                    .expect("parse date should not fail")
926            )
927        );
928        assert_eq!(meta.content_md5(), Some("fHcEH1vPwA6eTPqxuasXcg=="));
929        assert_eq!(meta.etag(), Some("CKWasoTgyPkCEAE="));
930        assert_eq!(meta.content_type(), Some("image/png"));
931        assert_eq!(meta.content_encoding(), Some("br"));
932        assert_eq!(meta.content_disposition(), Some("attachment"));
933        assert_eq!(meta.cache_control(), Some("public, max-age=3600"));
934        assert_eq!(meta.version(), Some("1660563214863653"));
935
936        let metadata = HashMap::from_iter([("location".to_string(), "everywhere".to_string())]);
937        assert_eq!(meta.user_metadata(), Some(&metadata));
938    }
939
940    #[test]
941    fn test_deserialize_list_response() {
942        let content = r#"
943    {
944  "kind": "storage#objects",
945  "prefixes": [
946    "dir/",
947    "test/"
948  ],
949  "items": [
950    {
951      "kind": "storage#object",
952      "id": "example/1.png/1660563214863653",
953      "selfLink": "https://www.googleapis.com/storage/v1/b/example/o/1.png",
954      "mediaLink": "https://content-storage.googleapis.com/download/storage/v1/b/example/o/1.png?generation=1660563214863653&alt=media",
955      "name": "1.png",
956      "bucket": "example",
957      "generation": "1660563214863653",
958      "metageneration": "1",
959      "contentType": "image/png",
960      "storageClass": "STANDARD",
961      "size": "56535",
962      "md5Hash": "fHcEH1vPwA6eTPqxuasXcg==",
963      "crc32c": "j/un9g==",
964      "etag": "CKWasoTgyPkCEAE=",
965      "timeCreated": "2022-08-15T11:33:34.866Z",
966      "updated": "2022-08-15T11:33:34.866Z",
967      "timeStorageClassUpdated": "2022-08-15T11:33:34.866Z"
968    },
969    {
970      "kind": "storage#object",
971      "id": "example/2.png/1660563214883337",
972      "selfLink": "https://www.googleapis.com/storage/v1/b/example/o/2.png",
973      "mediaLink": "https://content-storage.googleapis.com/download/storage/v1/b/example/o/2.png?generation=1660563214883337&alt=media",
974      "name": "2.png",
975      "bucket": "example",
976      "generation": "1660563214883337",
977      "metageneration": "1",
978      "contentType": "image/png",
979      "storageClass": "STANDARD",
980      "size": "45506",
981      "md5Hash": "e6LsGusU7pFJZk+114NV1g==",
982      "crc32c": "L00QAg==",
983      "etag": "CIm0s4TgyPkCEAE=",
984      "timeCreated": "2022-08-15T11:33:34.886Z",
985      "updated": "2022-08-15T11:33:34.886Z",
986      "timeStorageClassUpdated": "2022-08-15T11:33:34.886Z"
987    }
988  ]
989}
990    "#;
991
992        let output: ListResponse =
993            serde_json::from_str(content).expect("JSON deserialize must succeed");
994        assert!(output.next_page_token.is_none());
995        assert_eq!(output.items.len(), 2);
996        assert_eq!(output.items[0].name, "1.png");
997        assert_eq!(output.items[0].size, "56535");
998        assert_eq!(output.items[0].md5_hash, "fHcEH1vPwA6eTPqxuasXcg==");
999        assert_eq!(output.items[0].etag, "CKWasoTgyPkCEAE=");
1000        assert_eq!(output.items[0].updated, "2022-08-15T11:33:34.866Z");
1001        assert_eq!(output.items[1].name, "2.png");
1002        assert_eq!(output.items[1].size, "45506");
1003        assert_eq!(output.items[1].md5_hash, "e6LsGusU7pFJZk+114NV1g==");
1004        assert_eq!(output.items[1].etag, "CIm0s4TgyPkCEAE=");
1005        assert_eq!(output.items[1].updated, "2022-08-15T11:33:34.886Z");
1006        assert_eq!(output.items[1].content_type, "image/png");
1007        assert_eq!(output.prefixes, vec!["dir/", "test/"])
1008    }
1009
1010    #[test]
1011    fn test_deserialize_list_response_with_next_page_token() {
1012        let content = r#"
1013    {
1014  "kind": "storage#objects",
1015  "prefixes": [
1016    "dir/",
1017    "test/"
1018  ],
1019  "nextPageToken": "CgYxMC5wbmc=",
1020  "items": [
1021    {
1022      "kind": "storage#object",
1023      "id": "example/1.png/1660563214863653",
1024      "selfLink": "https://www.googleapis.com/storage/v1/b/example/o/1.png",
1025      "mediaLink": "https://content-storage.googleapis.com/download/storage/v1/b/example/o/1.png?generation=1660563214863653&alt=media",
1026      "name": "1.png",
1027      "bucket": "example",
1028      "generation": "1660563214863653",
1029      "metageneration": "1",
1030      "contentType": "image/png",
1031      "storageClass": "STANDARD",
1032      "size": "56535",
1033      "md5Hash": "fHcEH1vPwA6eTPqxuasXcg==",
1034      "crc32c": "j/un9g==",
1035      "etag": "CKWasoTgyPkCEAE=",
1036      "timeCreated": "2022-08-15T11:33:34.866Z",
1037      "updated": "2022-08-15T11:33:34.866Z",
1038      "timeStorageClassUpdated": "2022-08-15T11:33:34.866Z"
1039    },
1040    {
1041      "kind": "storage#object",
1042      "id": "example/2.png/1660563214883337",
1043      "selfLink": "https://www.googleapis.com/storage/v1/b/example/o/2.png",
1044      "mediaLink": "https://content-storage.googleapis.com/download/storage/v1/b/example/o/2.png?generation=1660563214883337&alt=media",
1045      "name": "2.png",
1046      "bucket": "example",
1047      "generation": "1660563214883337",
1048      "metageneration": "1",
1049      "contentType": "image/png",
1050      "storageClass": "STANDARD",
1051      "size": "45506",
1052      "md5Hash": "e6LsGusU7pFJZk+114NV1g==",
1053      "crc32c": "L00QAg==",
1054      "etag": "CIm0s4TgyPkCEAE=",
1055      "timeCreated": "2022-08-15T11:33:34.886Z",
1056      "updated": "2022-08-15T11:33:34.886Z",
1057      "timeStorageClassUpdated": "2022-08-15T11:33:34.886Z"
1058    }
1059  ]
1060}
1061    "#;
1062
1063        let output: ListResponse =
1064            serde_json::from_str(content).expect("JSON deserialize must succeed");
1065        assert_eq!(output.next_page_token, Some("CgYxMC5wbmc=".to_string()));
1066        assert_eq!(output.items.len(), 2);
1067        assert_eq!(output.items[0].name, "1.png");
1068        assert_eq!(output.items[0].size, "56535");
1069        assert_eq!(output.items[0].md5_hash, "fHcEH1vPwA6eTPqxuasXcg==");
1070        assert_eq!(output.items[0].etag, "CKWasoTgyPkCEAE=");
1071        assert_eq!(output.items[0].updated, "2022-08-15T11:33:34.866Z");
1072        assert_eq!(output.items[1].name, "2.png");
1073        assert_eq!(output.items[1].size, "45506");
1074        assert_eq!(output.items[1].md5_hash, "e6LsGusU7pFJZk+114NV1g==");
1075        assert_eq!(output.items[1].etag, "CIm0s4TgyPkCEAE=");
1076        assert_eq!(output.items[1].updated, "2022-08-15T11:33:34.886Z");
1077        assert_eq!(output.prefixes, vec!["dir/", "test/"])
1078    }
1079}
opendal/services/gcs/core.rs

opendal/services/gcs/
core.rs
