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

opendal/services/gcs/
core.rs
