Skip to content

feat(rest): wrap catalog responses in HttpResponse #2996

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
plusplusjiajia wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(rest): wrap catalog responses in HttpResponse #2996

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 11 additions & 1 deletion crates/catalog/rest/public-api.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,7 +126,7 @@ impl<'de> serde_core::de::Deserialize<'de> for iceberg_catalog_rest::ErrorRespon
pub fn iceberg_catalog_rest::ErrorResponse::deserialize<__D>(__deserializer: __D) -> core::result::Result<Self, <__D as serde_core::de::Deserializer>::Error> where __D: serde_core::de::Deserializer<'de>
pub struct iceberg_catalog_rest::HttpClient
impl iceberg_catalog_rest::HttpClient
pub async fn iceberg_catalog_rest::HttpClient::post_form(&self, url: &str, headers: &http::header::map::HeaderMap, form: &std::collections::hash::map::HashMap<&str, &str>) -> iceberg::error::Result<(http::status::StatusCode, alloc::vec::Vec<u8>)>
pub async fn iceberg_catalog_rest::HttpClient::post_form(&self, url: &str, headers: &http::header::map::HeaderMap, form: &std::collections::hash::map::HashMap<&str, &str>) -> iceberg::error::Result<iceberg_catalog_rest::HttpResponse>
pub fn iceberg_catalog_rest::HttpClient::with_auth_session(&self, auth_session: alloc::sync::Arc<dyn iceberg_catalog_rest::AuthSession>) -> Self
pub fn iceberg_catalog_rest::HttpClient::without_auth_session(&self) -> Self
impl core::clone::Clone for iceberg_catalog_rest::HttpClient
Expand All @@ -141,6 +141,16 @@ pub fn iceberg_catalog_rest::HttpRequest::headers_mut(&mut self) -> &mut http::h
pub fn iceberg_catalog_rest::HttpRequest::method(&self) -> &http::method::Method
pub fn iceberg_catalog_rest::HttpRequest::new(inner: reqwest::async_impl::request::Request) -> Self
pub fn iceberg_catalog_rest::HttpRequest::url_str(&self) -> &str
pub struct iceberg_catalog_rest::HttpResponse
impl iceberg_catalog_rest::HttpResponse
pub fn iceberg_catalog_rest::HttpResponse::body(&self) -> &[u8]
pub fn iceberg_catalog_rest::HttpResponse::headers(&self) -> &http::header::map::HeaderMap
pub fn iceberg_catalog_rest::HttpResponse::new(status: http::status::StatusCode, headers: http::header::map::HeaderMap, body: alloc::vec::Vec<u8>) -> Self
pub fn iceberg_catalog_rest::HttpResponse::status(&self) -> http::status::StatusCode
impl core::clone::Clone for iceberg_catalog_rest::HttpResponse
pub fn iceberg_catalog_rest::HttpResponse::clone(&self) -> iceberg_catalog_rest::HttpResponse
impl core::fmt::Debug for iceberg_catalog_rest::HttpResponse
pub fn iceberg_catalog_rest::HttpResponse::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result
pub struct iceberg_catalog_rest::ListNamespaceResponse
pub iceberg_catalog_rest::ListNamespaceResponse::namespaces: alloc::vec::Vec<iceberg::catalog::NamespaceIdent>
pub iceberg_catalog_rest::ListNamespaceResponse::next_page_token: core::option::Option<alloc::string::String>
Expand Down
12 changes: 7 additions & 5 deletions crates/catalog/rest/src/auth/oauth2.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -278,29 +278,31 @@ impl ClientCredentialsConfig {
.map(|(k, v)| (k.as_str(), v.as_str())),
);

let (status, body) = self
let response = self
.client
.post_form(&self.token_endpoint, &self.extra_headers, &params)
.await?;
let status = response.status();
let body = response.body();

let auth_res: TokenResponse = if status == StatusCode::OK {
Ok(serde_json::from_slice(&body).map_err(|e| {
Ok(serde_json::from_slice(body).map_err(|e| {
Error::new(
ErrorKind::Unexpected,
"Failed to parse response from rest catalog server!",
)
.with_context("operation", "auth")
.with_context("url", self.token_endpoint.clone())
.with_context("json", String::from_utf8_lossy(&body))
.with_context("json", String::from_utf8_lossy(body))
.with_source(e)
})?)
} else {
let e: ErrorResponse = serde_json::from_slice(&body).map_err(|e| {
let e: ErrorResponse = serde_json::from_slice(body).map_err(|e| {
Error::new(ErrorKind::Unexpected, "Received unexpected response")
.with_context("code", status.to_string())
.with_context("operation", "auth")
.with_context("url", self.token_endpoint.clone())
.with_context("json", String::from_utf8_lossy(&body))
.with_context("json", String::from_utf8_lossy(body))
.with_source(e)
})?;
Err(Error::from(e))
Expand Down
76 changes: 29 additions & 47 deletions crates/catalog/rest/src/catalog.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ use itertools::Itertools;
use reqwest::header::{
HeaderMap, HeaderName, HeaderValue, {self},
};
use reqwest::{Client, Method, Response, StatusCode, Url};
use reqwest::{Client, Method, StatusCode, Url};
use tokio::sync::OnceCell;
use typed_builder::TypedBuilder;

Expand All @@ -45,6 +45,7 @@ use crate::client::{
};
use crate::endpoint::{Endpoint, V1_NAMESPACE_EXISTS, V1_TABLE_EXISTS};
use crate::request::HttpRequest;
use crate::response::HttpResponse;
use crate::types::{
CatalogConfig, CommitTableRequest, CommitTableResponse, CreateNamespaceRequest,
CreateTableRequest, ListNamespaceResponse, ListTablesResponse, LoadTableResult,
Expand Down Expand Up @@ -339,7 +340,7 @@ impl RestCatalogConfig {

/// Merge the `RestCatalogConfig` with the a [`CatalogConfig`] (fetched from the REST server).
pub(crate) fn merge_with_config(mut self, mut config: CatalogConfig) -> Self {
if let Some(uri) = config.overrides.remove("uri") {
if let Some(uri) = config.overrides.remove(REST_CATALOG_PROP_URI) {
self.uri = uri;
}

Expand Down Expand Up @@ -479,7 +480,7 @@ impl RestClient {
}

/// Sends `request`, authenticated by the client's session.
async fn query_catalog(&self, request: HttpRequest) -> Result<Response> {
async fn query_catalog(&self, request: HttpRequest) -> Result<HttpResponse> {
self.http_client.query_catalog(request).await
}
}
Expand All @@ -488,7 +489,8 @@ impl RestClient {
#[derive(Debug)]
pub struct RestCatalog {
/// Injected through [`RestCatalogBuilder::with_auth_manager`]; otherwise
/// one is resolved from `rest.auth.type` when the context is built.
/// one is resolved from `rest.auth.type` when [`RestClient`] is
/// initialized.
auth_manager: Option<Arc<dyn AuthManager>>,
/// User config is stored as-is and never be changed.
///
Expand Down Expand Up @@ -545,8 +547,7 @@ impl RestCatalog {
_ => Err(deserialize_unexpected_catalog_error(
http_response,
client.http_client.disable_header_redaction(),
)
.await),
)),
}
}

Expand Down Expand Up @@ -691,8 +692,7 @@ impl RestCatalog {
_ => Err(deserialize_unexpected_catalog_error(
http_response,
client.http_client.disable_header_redaction(),
)
.await),
)),
}
}

Expand All @@ -714,12 +714,11 @@ impl RestCatalog {
let http_response = http_client.query_catalog(request).await?;

match http_response.status() {
StatusCode::OK => deserialize_catalog_response(http_response).await,
StatusCode::OK => deserialize_catalog_response(http_response),
_ => Err(deserialize_unexpected_catalog_error(
http_response,
http_client.disable_header_redaction(),
)
.await),
)),
}
}

Expand Down Expand Up @@ -766,7 +765,7 @@ impl RestCatalog {
}

/// All requests and expected responses are derived from the REST catalog API spec:
/// https://github.com/apache/iceberg/blob/main/open-api/rest-catalog-open-api.yaml
/// <https://github.com/apache/iceberg/blob/main/open-api/rest-catalog-open-api.yaml>
#[async_trait]
impl Catalog for RestCatalog {
async fn list_namespaces(
Expand Down Expand Up @@ -795,8 +794,7 @@ impl Catalog for RestCatalog {
match http_response.status() {
StatusCode::OK => {
let response =
deserialize_catalog_response::<ListNamespaceResponse>(http_response)
.await?;
deserialize_catalog_response::<ListNamespaceResponse>(http_response)?;

namespaces.extend(response.namespaces);

Expand All @@ -815,8 +813,7 @@ impl Catalog for RestCatalog {
return Err(deserialize_unexpected_catalog_error(
http_response,
client.http_client.disable_header_redaction(),
)
.await);
));
}
}
}
Expand Down Expand Up @@ -845,8 +842,7 @@ impl Catalog for RestCatalog {

match http_response.status() {
StatusCode::OK => {
let response =
deserialize_catalog_response::<NamespaceResponse>(http_response).await?;
let response = deserialize_catalog_response::<NamespaceResponse>(http_response)?;
Ok(Namespace::from(response))
}
StatusCode::CONFLICT => Err(Error::new(
Expand All @@ -856,8 +852,7 @@ impl Catalog for RestCatalog {
_ => Err(deserialize_unexpected_catalog_error(
http_response,
client.http_client.disable_header_redaction(),
)
.await),
)),
}
}

Expand All @@ -874,8 +869,7 @@ impl Catalog for RestCatalog {

match http_response.status() {
StatusCode::OK => {
let response =
deserialize_catalog_response::<NamespaceResponse>(http_response).await?;
let response = deserialize_catalog_response::<NamespaceResponse>(http_response)?;
Ok(Namespace::from(response))
}
StatusCode::NOT_FOUND => Err(Error::new(
Expand All @@ -885,8 +879,7 @@ impl Catalog for RestCatalog {
_ => Err(deserialize_unexpected_catalog_error(
http_response,
client.http_client.disable_header_redaction(),
)
.await),
)),
}
}

Expand Down Expand Up @@ -939,8 +932,7 @@ impl Catalog for RestCatalog {
_ => Err(deserialize_unexpected_catalog_error(
http_response,
client.http_client.disable_header_redaction(),
)
.await),
)),
}
}

Expand All @@ -962,7 +954,7 @@ impl Catalog for RestCatalog {
match http_response.status() {
StatusCode::OK => {
let response =
deserialize_catalog_response::<ListTablesResponse>(http_response).await?;
deserialize_catalog_response::<ListTablesResponse>(http_response)?;

identifiers.extend(response.identifiers);

Expand All @@ -981,8 +973,7 @@ impl Catalog for RestCatalog {
return Err(deserialize_unexpected_catalog_error(
http_response,
client.http_client.disable_header_redaction(),
)
.await);
));
}
}
}
Expand Down Expand Up @@ -1023,9 +1014,7 @@ impl Catalog for RestCatalog {
let http_response = client.query_catalog(request).await?;

let response = match http_response.status() {
StatusCode::OK => {
deserialize_catalog_response::<LoadTableResult>(http_response).await?
}
StatusCode::OK => deserialize_catalog_response::<LoadTableResult>(http_response)?,
StatusCode::NOT_FOUND => {
return Err(Error::new(
ErrorKind::NamespaceNotFound,
Expand All @@ -1042,8 +1031,7 @@ impl Catalog for RestCatalog {
return Err(deserialize_unexpected_catalog_error(
http_response,
client.http_client.disable_header_redaction(),
)
.await);
));
}
};

Expand Down Expand Up @@ -1096,7 +1084,7 @@ impl Catalog for RestCatalog {

let response = match http_response.status() {
StatusCode::OK | StatusCode::NOT_MODIFIED => {
deserialize_catalog_response::<LoadTableResult>(http_response).await?
deserialize_catalog_response::<LoadTableResult>(http_response)?
}
StatusCode::NOT_FOUND => {
return Err(Error::new(
Expand All @@ -1108,8 +1096,7 @@ impl Catalog for RestCatalog {
return Err(deserialize_unexpected_catalog_error(
http_response,
client.http_client.disable_header_redaction(),
)
.await);
));
}
};

Expand Down Expand Up @@ -1197,8 +1184,7 @@ impl Catalog for RestCatalog {
_ => Err(deserialize_unexpected_catalog_error(
http_response,
client.http_client.disable_header_redaction(),
)
.await),
)),
}
}

Expand Down Expand Up @@ -1228,9 +1214,7 @@ impl Catalog for RestCatalog {
let http_response = client.query_catalog(request).await?;

let response: LoadTableResult = match http_response.status() {
StatusCode::OK => {
deserialize_catalog_response::<LoadTableResult>(http_response).await?
}
StatusCode::OK => deserialize_catalog_response::<LoadTableResult>(http_response)?,
StatusCode::NOT_FOUND => {
return Err(Error::new(
ErrorKind::NamespaceNotFound,
Expand All @@ -1247,8 +1231,7 @@ impl Catalog for RestCatalog {
return Err(deserialize_unexpected_catalog_error(
http_response,
client.http_client.disable_header_redaction(),
)
.await);
));
}
};

Expand Down Expand Up @@ -1291,7 +1274,7 @@ impl Catalog for RestCatalog {
let http_response = client.query_catalog(request).await?;

let response: CommitTableResponse = match http_response.status() {
StatusCode::OK => deserialize_catalog_response(http_response).await?,
StatusCode::OK => deserialize_catalog_response(http_response)?,
StatusCode::NOT_FOUND => {
return Err(Error::new(
ErrorKind::TableNotFound,
Expand Down Expand Up @@ -1327,8 +1310,7 @@ impl Catalog for RestCatalog {
return Err(deserialize_unexpected_catalog_error(
http_response,
client.http_client.disable_header_redaction(),
)
.await);
));
}
};

Expand Down
Loading
Loading