lakestream 0.0.3

Portable file-utility for object-stores
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

use std::collections::HashMap;

use url::form_urlencoded;

use super::client::S3Client;
use crate::{LakestreamError, AWS_MAX_LIST_OBJECTS};

pub trait Headers {
    fn generate_list_buckets_headers(
        &self,
    ) -> Result<HashMap<String, String>, LakestreamError>;
    fn generate_list_objects_headers(
        &mut self,
        prefix: Option<&str>,
        max_keys: Option<u32>,
        continuation_token: Option<&str>,
    ) -> Result<HashMap<String, String>, LakestreamError>;
    fn generate_get_object_headers(
        &mut self,
        object_key: &str,
    ) -> Result<HashMap<String, String>, LakestreamError>;
    fn create_list_objects_query_string(
        &self,
        prefix: Option<&str>,
        max_keys: Option<u32>,
        continuation_token: Option<&str>,
    ) -> String;
}

impl Headers for S3Client {
    fn generate_list_buckets_headers(
        &self,
    ) -> Result<HashMap<String, String>, LakestreamError> {
        let method = "GET";
        self.request_builder.generate_headers(
            self.config(),
            method,
            None,
            None,
            None,
        )
    }

    fn generate_list_objects_headers(
        &mut self,
        prefix: Option<&str>,
        max_keys: Option<u32>,
        continuation_token: Option<&str>,
    ) -> Result<HashMap<String, String>, LakestreamError> {
        let method = "GET";
        let query_string = Some(self.create_list_objects_query_string(
            prefix,
            max_keys,
            continuation_token,
        ));

        self.query_string = query_string.clone();
        self.request_builder.generate_headers(
            self.config(),
            method,
            self.resource.as_deref(),
            query_string.as_deref(),
            None,
        )
    }

    fn generate_get_object_headers(
        &mut self,
        object_key: &str,
    ) -> Result<HashMap<String, String>, LakestreamError> {
        self.resource = Some(object_key.to_string());
        let method = "GET";
        self.request_builder.generate_headers(
            self.config(),
            method,
            self.resource.as_deref(),
            self.query_string.as_deref(),
            None,
        )
    }

    fn create_list_objects_query_string(
        &self,
        prefix: Option<&str>,
        max_keys: Option<u32>,
        continuation_token: Option<&str>,
    ) -> String {
        // Ensure max_keys does not exceed AWS_MAX_LIST_OBJECTS
        let max_keys = max_keys
            .map(|keys| std::cmp::min(keys, AWS_MAX_LIST_OBJECTS))
            .unwrap_or(AWS_MAX_LIST_OBJECTS);

        let mut query_parts = form_urlencoded::Serializer::new(String::new());
        query_parts.append_pair("list-type", "2");
        query_parts.append_pair("max-keys", &max_keys.to_string());
        query_parts.append_pair("delimiter", "/");
        query_parts.append_pair("encoding-type", "url");

        if let Some(p) = prefix {
            query_parts.append_pair("prefix", p);
        }
        if let Some(token) = continuation_token {
            query_parts.append_pair("continuation-token", token);
        }

        query_parts.finish()
    }
}