add list_namespaces

Author: flaneur2020

crates/catalog/s3tables/src/catalog.rs

@@ -2,27 +2,35 @@ use std::collections::HashMap;
 
 use anyhow::anyhow;
 use async_trait::async_trait;
-use aws_config::BehaviorVersion;
 use iceberg::table::Table;
 use iceberg::{
     Catalog, Error, ErrorKind, Namespace, NamespaceIdent, Result, TableCommit, TableCreation,
     TableIdent,
 };
 
+use crate::utils::create_sdk_config;
+
+#[derive(Debug)]
+pub struct S3TablesCatalogConfig {
+    table_bucket_arn: String,
+    properties: HashMap<String, String>,
+    endpoint_url: Option<String>,
+}
+
 /// S3Tables catalog implementation.
 #[derive(Debug)]
 pub struct S3TablesCatalog {
-    table_bucket_arn: String,
-    client: aws_sdk_s3tables::Client,
+    config: S3TablesCatalogConfig,
+    s3tables_client: aws_sdk_s3tables::Client,
 }
 
 impl S3TablesCatalog {
-    pub async fn new(table_bucket_arn: String) -> Self {
-        let config = aws_config::load_defaults(BehaviorVersion::latest()).await;
-        let client = aws_sdk_s3tables::Client::new(&config);
+    pub async fn new(config: S3TablesCatalogConfig) -> Self {
+        let aws_config = create_sdk_config(&config.properties, config.endpoint_url.clone()).await;
+        let s3tables_client = aws_sdk_s3tables::Client::new(&aws_config);
         Self {
-            table_bucket_arn,
-            client,
+            config,
+            s3tables_client,
         }
     }
 }
@@ -34,21 +42,16 @@ impl Catalog for S3TablesCatalog {
         parent: Option<&NamespaceIdent>,
     ) -> Result<Vec<NamespaceIdent>> {
         let mut req = self
-            .client
+            .s3tables_client
             .list_namespaces()
-            .table_bucket_arn(self.table_bucket_arn.clone());
+            .table_bucket_arn(self.config.table_bucket_arn.clone());
         if let Some(parent) = parent {
             req = req.prefix(parent.to_url_string());
         }
         let resp = req.send().await.map_err(from_aws_sdk_error)?;
         let mut result = Vec::new();
         for ns in resp.namespaces() {
-            let ns_names = ns.namespace();
-            result.extend(
-                ns_names
-                    .into_iter()
-                    .map(|name| NamespaceIdent::new(name.to_string())),
-            );
+            result.push(NamespaceIdent::from_vec(ns.namespace().to_vec())?);
         }
         Ok(result)
     }

crates/catalog/s3tables/src/lib.rs

@@ -20,3 +20,4 @@
 #![deny(missing_docs)]
 
 mod catalog;
+mod utils;

crates/catalog/s3tables/src/utils.rs

@@ -0,0 +1,55 @@
+use std::collections::HashMap;
+
+use aws_config::default_provider::endpoint_url;
+use aws_config::{BehaviorVersion, Region, SdkConfig};
+use aws_sdk_s3tables::config::Credentials;
+
+/// Property aws profile name
+pub const AWS_PROFILE_NAME: &str = "profile_name";
+/// Property aws region
+pub const AWS_REGION_NAME: &str = "region_name";
+/// Property aws access key
+pub const AWS_ACCESS_KEY_ID: &str = "aws_access_key_id";
+/// Property aws secret access key
+pub const AWS_SECRET_ACCESS_KEY: &str = "aws_secret_access_key";
+/// Property aws session token
+pub const AWS_SESSION_TOKEN: &str = "aws_session_token";
+
+/// Creates an aws sdk configuration based on
+/// provided properties and an optional endpoint URL.
+pub(crate) async fn create_sdk_config(
+    properties: &HashMap<String, String>,
+    endpoint_url: Option<String>,
+) -> SdkConfig {
+    let mut config = aws_config::defaults(BehaviorVersion::latest());
+
+    if properties.is_empty() {
+        return config.load().await;
+    }
+
+    if let Some(endpoint_url) = endpoint_url {
+        config = config.endpoint_url(endpoint_url);
+    }
+
+    if let (Some(access_key), Some(secret_key)) = (
+        properties.get(AWS_ACCESS_KEY_ID),
+        properties.get(AWS_SECRET_ACCESS_KEY),
+    ) {
+        let session_token = properties.get(AWS_SESSION_TOKEN).cloned();
+        let credentials_provider =
+            Credentials::new(access_key, secret_key, session_token, None, "properties");
+
+        config = config.credentials_provider(credentials_provider)
+    };
+
+    if let Some(profile_name) = properties.get(AWS_PROFILE_NAME) {
+        config = config.profile_name(profile_name);
+    }
+
+    if let Some(region_name) = properties.get(AWS_REGION_NAME) {
+        let region = Region::new(region_name.clone());
+        config = config.region(region);
+    }
+
+    config.load().await
+}