iceberg handlers modularized, api locked for 0.3.0

Author: AlexMercedCoder

pangolin/pangolin_api/src/asset_handlers.rs

@@ -13,7 +13,8 @@ use crate::auth::TenantId;
 use crate::authz::check_permission;
 use pangolin_core::permission::{PermissionScope, Action};
 use pangolin_core::user::UserSession;
-use crate::iceberg_handlers::{AppState, parse_table_identifier};
+use crate::iceberg::AppState;
+use crate::iceberg::parse_table_identifier;
 use utoipa::ToSchema;
 
 #[derive(Deserialize, ToSchema)]
@@ -320,7 +321,7 @@ pub struct AssetSummary {
     pub name: String,
     pub namespace: Vec<String>,
     pub kind: AssetType,
-    pub identifier: crate::iceberg_handlers::TableIdentifier,
+    pub identifier: crate::iceberg::TableIdentifier,
 }
 
 /// List all assets in a namespace (Iceberg tables + Generic assets)
@@ -406,7 +407,7 @@ pub async fn list_assets(
                     name: asset.name.clone(),
                     namespace: namespace_parts.clone(),
                     kind: asset.kind,
-                    identifier: crate::iceberg_handlers::TableIdentifier {
+                    identifier: crate::iceberg::TableIdentifier {
                         namespace: namespace_parts.clone(),
                         name: asset.name,
                     }
@@ -422,7 +423,7 @@ pub async fn list_assets(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::iceberg_handlers::AppState;
+    use crate::iceberg::AppState;
     use pangolin_core::user::UserSession;
     use pangolin_core::permission::{PermissionScope, Action};
     use crate::tenant::TenantId;

pangolin/pangolin_api/src/business_metadata_handlers.rs

@@ -12,7 +12,7 @@ use pangolin_core::business_metadata::{BusinessMetadata, AccessRequest, RequestS
 use pangolin_core::user::{UserSession, UserRole};
 use pangolin_core::permission::{Action, PermissionScope};
 use uuid::Uuid;
-use crate::iceberg_handlers::AppState;
+use crate::iceberg::AppState;
 use utoipa::ToSchema;
 
 #[derive(Deserialize, Serialize, ToSchema)]

pangolin/pangolin_api/src/dashboard_handlers.rs

@@ -7,7 +7,7 @@ use axum::{
 use serde::{Deserialize, Serialize};
 use pangolin_store::CatalogStore;
 use pangolin_core::user::{UserSession, UserRole};
-use crate::iceberg_handlers::AppState;
+use crate::iceberg::AppState;
 use crate::error::ApiError;
 use utoipa::ToSchema;
 

pangolin/pangolin_api/src/iceberg/config.rs

@@ -0,0 +1,39 @@
+use axum::Json;
+use std::collections::HashMap;
+use super::types::CatalogConfig;
+
+/// Get Iceberg Catalog Configuration
+#[utoipa::path(
+    get,
+    path = "/v1/config",
+    tag = "Iceberg REST",
+    responses(
+        (status = 200, description = "Catalog configuration", body = CatalogConfig),
+    )
+)]
+pub async fn get_iceberg_catalog_config_handler() -> Json<CatalogConfig> {
+    // Return Iceberg REST catalog config
+    // Use X-Iceberg-Access-Delegation header to enable credential vending
+    let mut defaults = HashMap::new();
+    
+    // This header tells PyIceberg to request credentials via the vend-credentials endpoint
+    // PyIceberg will call POST /v1/{prefix}/namespaces/{namespace}/tables/{table}/credentials
+    defaults.insert("header.X-Iceberg-Access-Delegation".to_string(), "vended-credentials".to_string());
+    
+    // Optionally provide S3 endpoint if using MinIO or custom S3
+    if let Ok(endpoint) = std::env::var("S3_ENDPOINT") {
+        defaults.insert("s3.endpoint".to_string(), endpoint);
+    } else if let Ok(endpoint) = std::env::var("AWS_ENDPOINT_URL") {
+        defaults.insert("s3.endpoint".to_string(), endpoint);
+    }
+    
+    // Add region if specified
+    if let Ok(region) = std::env::var("AWS_REGION") {
+        defaults.insert("s3.region".to_string(), region);
+    }
+    
+    Json(CatalogConfig {
+        defaults,
+        overrides: HashMap::new(),
+    })
+}

pangolin/pangolin_api/src/iceberg/mod.rs

@@ -0,0 +1,59 @@
+use axum::{
+    body::Body,
+    http::{HeaderMap, Method, StatusCode},
+    response::IntoResponse,
+    Json,
+};
+use bytes::Bytes;
+use pangolin_store::CatalogStore;
+use pangolin_core::model::CatalogType;
+use std::sync::Arc;
+use crate::federated_proxy::FederatedCatalogProxy;
+
+pub mod config;
+pub mod namespaces;
+pub mod tables;
+pub mod types;
+
+// Re-export types for convenience
+pub use types::*;
+pub type AppState = std::sync::Arc<dyn pangolin_store::CatalogStore + Send + Sync>;
+
+/// Helper function to check if a catalog is federated and forward the request if so
+pub async fn check_and_forward_if_federated(
+    store: &Arc<dyn CatalogStore + Send + Sync>,
+    tenant_id: uuid::Uuid,
+    catalog_name: &str,
+    method: Method,
+    path: &str,
+    body: Option<Bytes>,
+    headers: HeaderMap,
+) -> Option<axum::response::Response> {
+    // Get the catalog
+    let catalog = match store.get_catalog(tenant_id, catalog_name.to_string()).await {
+        Ok(Some(c)) => c,
+        Ok(None) => return None, // Catalog not found, let handler deal with it
+        Err(_) => return None,
+    };
+
+    // Check if it's federated
+    if catalog.catalog_type == CatalogType::Federated {
+        if let Some(config) = catalog.federated_config {
+            let proxy = FederatedCatalogProxy::new();
+            match proxy.forward_request(&config, method, path, body, headers).await {
+                Ok(response) => Some(response),
+                Err(e) => Some((
+                    StatusCode::BAD_GATEWAY,
+                    Json(serde_json::json!({"error": format!("Federated catalog error: {}", e)})),
+                ).into_response()),
+            }
+        } else {
+            Some((
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(serde_json::json!({"error": "Federated catalog missing configuration"})),
+            ).into_response())
+        }
+    } else {
+        None // Not federated, continue with local handling
+    }
+}