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

opendal_core/services/gcs/
core.rs
