Skip to content

refactor: replace FromResponse trait with fetch_inner + fetch_conditi… #23

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

Merged
gruuya merged 2 commits into from
Jul 15, 2026
Merged

refactor: replace FromResponse trait with fetch_inner + fetch_conditi… #23

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
54ef976
refactor: replace FromResponse trait with fetch_inner + fetch_conditi…
JanKaul Jul 15, 2026
94be523
refactor: split load_table into load_table + load_table_conditional
JanKaul Jul 15, 2026
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
66 changes: 49 additions & 17 deletions catalogs/iceberg-rest-catalog/src/apis/catalog_api_api.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -444,7 +444,7 @@ pub async fn commit_transaction(
let uri_str = "transactions/commit".to_string();
let method = reqwest::Method::POST;

fetch::fetch_empty(
fetch::fetch(
configuration,
method,
prefix,
Expand Down Expand Up @@ -548,7 +548,7 @@ pub async fn drop_namespace(

let method = reqwest::Method::DELETE;

fetch::fetch_empty(configuration, method, prefix, &uri_str, &(), None, None).await
fetch::fetch(configuration, method, prefix, &uri_str, &(), None, None).await
}

/// Remove a table from the catalog
Expand All @@ -571,7 +571,7 @@ pub async fn drop_table(
);
let method = reqwest::Method::DELETE;

fetch::fetch_empty(
fetch::fetch(
configuration,
method,
prefix,
Expand All @@ -597,7 +597,7 @@ pub async fn drop_view(
);
let method = reqwest::Method::DELETE;

fetch::fetch_empty(configuration, method, prefix, &uri_str, &(), None, None).await
fetch::fetch(configuration, method, prefix, &uri_str, &(), None, None).await
}

/// List all namespaces at a certain level, optionally starting from a given parent namespace. If table accounting.tax.paid.info exists, using 'SELECT NAMESPACE IN accounting' would translate into `GET /namespaces?parent=accounting` and must return a namespace, [\"accounting\", \"tax\"] only. Using 'SELECT NAMESPACE IN accounting.tax' would translate into `GET /namespaces?parent=accounting%1Ftax` and must return a namespace, [\"accounting\", \"tax\", \"paid\"]. If `parent` is not provided, all top-level namespaces should be listed.
Expand Down Expand Up @@ -783,19 +783,16 @@ pub async fn load_namespace_metadata(
/// Load a table from the catalog. The response contains the table's full metadata plus any extra
/// configuration to apply to it, which may include credentials to use for subsequent requests.
///
/// The headers argument can control how the Iceberg catalog handles credentials (via
/// `X-Iceberg-Access-Delegation`) and whether to use HTTP cache semantics (via `If-None-Match`).
pub async fn load_table<T>(
/// The headers argument can control how the Iceberg catalog handles credentials via
/// `X-Iceberg-Access-Delegation`. For HTTP cache semantics use [`load_table_conditional`].
pub async fn load_table(
Comment thread
gruuya marked this conversation as resolved.
configuration: &configuration::Configuration,
prefix: Option<&str>,
namespace: &str,
table: &str,
headers: HashMap<String, String>,
snapshots: Option<&str>,
) -> Result<T, Error<LoadTableError>>
where
T: super::FromResponse<LoadTableError>,
{
) -> Result<models::LoadTableResult, Error<LoadTableError>> {
let mut query_params = HashMap::new();
if let Some(snapshots) = snapshots {
query_params.insert("snapshots".to_owned(), snapshots.to_string());
Expand All @@ -819,6 +816,41 @@ where
.await
}

/// Load a table from the catalog with HTTP cache semantics. Pass an `If-None-Match` header with
/// a previously received ETag to skip re-fetching unchanged metadata; the server returns
/// `304 Not Modified` in that case and this function returns [`super::Conditional::NotModified`].
/// Use [`load_table`] when caching is not needed.
pub async fn load_table_conditional(
configuration: &configuration::Configuration,
prefix: Option<&str>,
namespace: &str,
table: &str,
headers: HashMap<String, String>,
snapshots: Option<&str>,
) -> Result<super::Conditional<models::LoadTableResult>, Error<LoadTableError>> {
let mut query_params = HashMap::new();
if let Some(snapshots) = snapshots {
query_params.insert("snapshots".to_owned(), snapshots.to_string());
}

let uri_str = format!(
"namespaces/{namespace}/tables/{table}",
namespace = crate::apis::urlencode(namespace),
table = crate::apis::urlencode(table)
);

fetch::fetch_conditional(
configuration,
reqwest::Method::GET,
prefix,
&uri_str,
&(),
Some(headers),
Some(query_params),
)
.await
}

/// Load a view from the catalog. The response contains both configuration and view metadata. The configuration, if non-empty is used as additional configuration for the view that overrides catalog configuration. The response also contains the view's full metadata, matching the view metadata JSON file. The catalog configuration may contain credentials that should be used for subsequent requests for the view. The configuration key \"token\" is used to pass an access token to be used as a bearer token for view requests. Otherwise, a token may be passed using a RFC 8693 token type as a configuration key. For example, \"urn:ietf:params:oauth:token-type:jwt=<JWT-token>\".
pub async fn load_view(
configuration: &configuration::Configuration,
Expand Down Expand Up @@ -849,7 +881,7 @@ pub async fn namespace_exists(

let method = reqwest::Method::HEAD;

fetch::fetch_empty(configuration, method, prefix, &uri_str, &(), None, None).await
fetch::fetch(configuration, method, prefix, &uri_str, &(), None, None).await
}

/// Register a table using given metadata file location.
Expand Down Expand Up @@ -887,7 +919,7 @@ pub async fn rename_table(
let uri_str = "tables/rename".to_string();
let method = reqwest::Method::POST;

fetch::fetch_empty(
fetch::fetch(
configuration,
method,
prefix,
Expand All @@ -908,7 +940,7 @@ pub async fn rename_view(
let uri_str = "views/rename".to_string();
let method = reqwest::Method::POST;

fetch::fetch_empty(
fetch::fetch(
configuration,
method,
prefix,
Expand Down Expand Up @@ -961,7 +993,7 @@ pub async fn report_metrics(
);
let method = reqwest::Method::POST;

fetch::fetch_empty(
fetch::fetch(
configuration,
method,
prefix,
Expand All @@ -988,7 +1020,7 @@ pub async fn table_exists(

let method = reqwest::Method::HEAD;

fetch::fetch_empty(configuration, method, prefix, &uri_str, &(), None, None).await
fetch::fetch(configuration, method, prefix, &uri_str, &(), None, None).await
}

/// Set and/or remove properties on a namespace. The request body specifies a list of properties to remove and a map of key value pairs to update. Properties that are not in the request are not modified or removed by this call. Server implementations are not required to support namespace properties.
Expand Down Expand Up @@ -1058,5 +1090,5 @@ pub async fn view_exists(

let method = reqwest::Method::HEAD;

fetch::fetch_empty(configuration, method, prefix, &uri_str, &(), None, None).await
fetch::fetch(configuration, method, prefix, &uri_str, &(), None, None).await
}
142 changes: 58 additions & 84 deletions catalogs/iceberg-rest-catalog/src/apis/fetch.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,87 +2,19 @@ use crate::apis::{configuration, ResponseContent};

use super::{Conditional, Error};

use async_trait::async_trait;
use std::collections::HashMap;

/// Turns a raw [`reqwest::Response`] into the caller's chosen output type. The output type
/// selects how the response is interpreted, so a single [`fetch`] serves every case:
/// - a `Deserialize` type parses the JSON body (an empty body is treated as `null`, so `()`
/// works for endpoints with no content);
/// - [`Conditional<T>`] additionally honours `304 Not Modified` and captures the response `ETag`.
#[async_trait]
pub trait FromResponse<E>: Sized {
async fn from_response(resp: reqwest::Response) -> Result<Self, Error<E>>;
}

#[async_trait]
impl<T, E> FromResponse<E> for T
where
T: serde::de::DeserializeOwned,
E: serde::de::DeserializeOwned,
{
async fn from_response(resp: reqwest::Response) -> Result<Self, Error<E>> {
let status = resp.status();
let content = resp.text().await?;
if !status.is_client_error() && !status.is_server_error() {
let body = if content.is_empty() { "null" } else { &content };
serde_json::from_str(body).map_err(Error::from)
} else {
let entity: Option<E> = serde_json::from_str(&content).ok();
Err(Error::ResponseError(ResponseContent {
status,
content,
entity,
}))
}
}
}

#[async_trait]
impl<T, E> FromResponse<E> for Conditional<T>
where
T: serde::de::DeserializeOwned,
E: serde::de::DeserializeOwned,
{
async fn from_response(resp: reqwest::Response) -> Result<Self, Error<E>> {
if resp.status() == reqwest::StatusCode::NOT_MODIFIED {
return Ok(Conditional::NotModified);
}
// Capture the ETag before consuming the response body.
let etag = resp
.headers()
.get(reqwest::header::ETAG)
.and_then(|v| v.to_str().ok())
.map(str::to_owned);
let status = resp.status();
let content = resp.text().await?;
if !status.is_client_error() && !status.is_server_error() {
let body = if content.is_empty() { "null" } else { &content };
let value = serde_json::from_str(body).map_err(Error::from)?;
Ok(Conditional::Modified { value, etag })
} else {
let entity: Option<E> = serde_json::from_str(&content).ok();
Err(Error::ResponseError(ResponseContent {
status,
content,
entity,
}))
}
}
}

pub(crate) async fn fetch<R, T, E>(
pub(crate) async fn fetch_inner<R, E>(
configuration: &configuration::Configuration,
method: reqwest::Method,
prefix: Option<&str>,
uri_str: &str,
request: &R,
headers: Option<HashMap<String, String>>,
query_params: Option<HashMap<String, String>>,
) -> Result<T, Error<E>>
) -> Result<reqwest::Response, Error<E>>
where
R: serde::Serialize + ?Sized,
T: FromResponse<E>,
{
let uri_base = build_uri_base(configuration, prefix);
let client = &configuration.client;
Expand Down Expand Up @@ -135,33 +67,75 @@ where
}

let req = req_builder.build()?;
let resp = client.execute(req).await?;
T::from_response(resp).await
client.execute(req).await.map_err(Error::from)
}

pub(crate) async fn fetch_empty<R, E>(
pub(crate) async fn fetch<R, T, E>(
configuration: &configuration::Configuration,
method: reqwest::Method,
prefix: Option<&str>,
uri_str: &str,
request: &R,
headers: Option<HashMap<String, String>>,
query_params: Option<HashMap<String, String>>,
) -> Result<T, Error<E>>
where
R: serde::Serialize + ?Sized,
T: serde::de::DeserializeOwned,
E: serde::de::DeserializeOwned,
{
let resp = fetch_inner(configuration, method, prefix, uri_str, request, headers, query_params).await?;
let status = resp.status();
let content = resp.text().await?;
if !status.is_client_error() && !status.is_server_error() {
let body = if content.is_empty() { "null" } else { &content };
serde_json::from_str(body).map_err(Error::from)
} else {
let entity: Option<E> = serde_json::from_str(&content).ok();
Err(Error::ResponseError(ResponseContent {
status,
content,
entity,
}))
}
}

pub(crate) async fn fetch_conditional<R, T, E>(
configuration: &configuration::Configuration,
method: reqwest::Method,
prefix: Option<&str>,
uri_str: &str,
request: &R,
headers: Option<HashMap<String, String>>,
query_params: Option<HashMap<String, String>>,
) -> Result<(), Error<E>>
) -> Result<Conditional<T>, Error<E>>
where
R: serde::Serialize + ?Sized,
T: serde::de::DeserializeOwned,
E: serde::de::DeserializeOwned,
{
fetch(
configuration,
method,
prefix,
uri_str,
request,
headers,
query_params,
)
.await
let resp = fetch_inner(configuration, method, prefix, uri_str, request, headers, query_params).await?;
if resp.status() == reqwest::StatusCode::NOT_MODIFIED {
return Ok(Conditional::NotModified);
}
let resp_headers = resp.headers().clone();
let status = resp.status();
let content = resp.text().await?;
if !status.is_client_error() && !status.is_server_error() {
let body = if content.is_empty() { "null" } else { &content };
let value = serde_json::from_str(body).map_err(Error::from)?;
Ok(Conditional::Modified {
value,
headers: resp_headers,
})
} else {
let entity: Option<E> = serde_json::from_str(&content).ok();
Err(Error::ResponseError(ResponseContent {
status,
content,
entity,
}))
}
}

/// Build the base URI for REST API calls with proper prefix encoding
Expand Down
8 changes: 5 additions & 3 deletions catalogs/iceberg-rest-catalog/src/apis/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,11 @@ pub struct ResponseContent<T> {
pub enum Conditional<T> {
/// Server returned `304 Not Modified`; the caller's cached value is still current.
NotModified,
/// Server returned a fresh body, along with its `ETag` if one was present.
Modified { value: T, etag: Option<String> },
/// Server returned a fresh body along with all HTTP response headers.
Modified {
value: T,
headers: reqwest::header::HeaderMap,
},
}

#[derive(Debug)]
Expand Down Expand Up @@ -114,6 +117,5 @@ pub mod configuration_api_api;
pub(crate) mod fetch;
pub mod o_auth2_api_api;

pub use fetch::FromResponse;

pub mod configuration;