iceberg_catalog_rest/
request.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18//! The request type REST catalog authentication works with.
19
20use http::{HeaderMap, Method};
21use iceberg::Result;
22use reqwest::{Request, RequestBuilder};
23
24/// An outgoing REST request being authenticated by an
25/// [`AuthSession`](crate::AuthSession).
26///
27/// Wraps the request so a session mutates it through the stable
28/// `http` crate types rather than the concrete request type the REST catalog
29/// uses internally.
30pub struct HttpRequest {
31    inner: Request,
32}
33
34impl HttpRequest {
35    /// Wraps a request, e.g. to unit-test a custom
36    /// [`AuthSession`](crate::AuthSession).
37    pub fn new(inner: Request) -> Self {
38        Self { inner }
39    }
40
41    /// Builds the request `builder` describes.
42    pub(crate) fn build(builder: RequestBuilder) -> Result<Self> {
43        Ok(Self::new(builder.build()?))
44    }
45
46    /// The wrapped request, for the client that sends it.
47    pub(crate) fn into_inner(self) -> Request {
48        self.inner
49    }
50
51    /// The request method.
52    pub fn method(&self) -> &Method {
53        self.inner.method()
54    }
55
56    /// The request URL, as a string (scheme, host, path and query).
57    pub fn url_str(&self) -> &str {
58        self.inner.url().as_str()
59    }
60
61    /// The request headers.
62    pub fn headers(&self) -> &HeaderMap {
63        self.inner.headers()
64    }
65
66    /// The mutable request headers, e.g. to add an `Authorization` header.
67    pub fn headers_mut(&mut self) -> &mut HeaderMap {
68        self.inner.headers_mut()
69    }
70
71    /// The request body, distinguishing an absent body from a streaming one:
72    /// signers can sign [`HttpRequestBody::Empty`] (empty-payload hash) and
73    /// [`HttpRequestBody::Buffered`], but not [`HttpRequestBody::Streaming`].
74    pub fn body(&self) -> HttpRequestBody<'_> {
75        match self.inner.body() {
76            None => HttpRequestBody::Empty,
77            Some(body) => match body.as_bytes() {
78                Some(bytes) => HttpRequestBody::Buffered(bytes),
79                None => HttpRequestBody::Streaming,
80            },
81        }
82    }
83}
84
85/// The body of an [`HttpRequest`], as seen by authentication.
86#[derive(Debug, Clone, Copy, PartialEq, Eq)]
87pub enum HttpRequestBody<'a> {
88    /// No body is set.
89    Empty,
90    /// An in-memory body.
91    Buffered(&'a [u8]),
92    /// A streaming body, whose bytes are not available for e.g. signing.
93    Streaming,
94}
95
96impl<'a> HttpRequestBody<'a> {
97    /// The signable bytes: empty for [`Self::Empty`], the buffer for
98    /// [`Self::Buffered`], and `None` for [`Self::Streaming`].
99    pub fn as_bytes(&self) -> Option<&'a [u8]> {
100        match self {
101            HttpRequestBody::Empty => Some(&[]),
102            HttpRequestBody::Buffered(bytes) => Some(bytes),
103            HttpRequestBody::Streaming => None,
104        }
105    }
106}
107
108#[cfg(test)]
109mod tests {
110    use reqwest::Client;
111
112    use super::*;
113
114    #[test]
115    fn test_http_request_body_states() {
116        let client = Client::new();
117
118        // No body at all.
119        let req = client
120            .get("https://rest.example.com/v1/config")
121            .build()
122            .unwrap();
123        let http_req = HttpRequest::new(req);
124        let body = http_req.body();
125        assert_eq!(body, HttpRequestBody::Empty);
126        assert_eq!(body.as_bytes(), Some(&[] as &[u8]));
127
128        // An in-memory body.
129        let req = client
130            .post("https://rest.example.com/v1/namespaces")
131            .body("{}")
132            .build()
133            .unwrap();
134        let http_req = HttpRequest::new(req);
135        let body = http_req.body();
136        assert_eq!(body, HttpRequestBody::Buffered(b"{}"));
137        assert_eq!(body.as_bytes(), Some(b"{}" as &[u8]));
138
139        // A streaming body: bytes are unavailable, so it must not sign as empty.
140        let req = client
141            .post("https://rest.example.com/v1/namespaces")
142            .body(reqwest::Body::wrap_stream(futures::stream::once(async {
143                Ok::<_, std::io::Error>(bytes::Bytes::from_static(b"chunk"))
144            })))
145            .build()
146            .unwrap();
147        let http_req = HttpRequest::new(req);
148        let body = http_req.body();
149        assert_eq!(body, HttpRequestBody::Streaming);
150        assert_eq!(body.as_bytes(), None);
151    }
152}
iceberg_catalog_rest/request.rs

iceberg_catalog_rest/
request.rs
