tessera/
client.rs

1//! Synchronous Tessera API client.
2
3#[cfg(feature = "polars")]
4use polars::prelude::{DataFrame, LazyFrame};
5
6use crate::base::{
7    PreparedRequest, datasets_request, download_request, parse_retry_after, partitions_request,
8    should_retry,
9};
10use crate::config::{ClientConfig, backoff_delay};
11use crate::error::{TesseraError, error_from_response};
12use crate::models::{
13    DatasetsResponse, DownloadResponse, IntoCoins, IntoMonths, PartitionRef, PartitionsResponse,
14};
15#[cfg(any(feature = "polars", feature = "duckdb"))]
16use crate::readers::ResolvedPartition;
17use crate::readers::expand_refs;
18#[cfg(any(feature = "polars", feature = "duckdb"))]
19use crate::resolver::resolve_sync;
20
21/// A synchronous client for the Tessera API.
22///
23/// Call [`TesseraClient::new`] with an explicit API key or rely on
24/// `$TESSERA_API_KEY`.
25///
26/// # Panics
27///
28/// The constructor builds a private tokio runtime (needed for Polars cloud
29/// reads), so a `TesseraClient` must not be constructed or used from inside
30/// another async runtime — use [`crate::AsyncTesseraClient`] there instead.
31pub struct TesseraClient {
32    config: ClientConfig,
33    http: reqwest::blocking::Client,
34    #[cfg(feature = "polars")]
35    runtime: tokio::runtime::Runtime,
36}
37
38impl TesseraClient {
39    /// Create a client with defaults, resolving the key from `api_key` or
40    /// `$TESSERA_API_KEY`.
41    ///
42    /// # Errors
43    ///
44    /// Returns [`TesseraError::Configuration`] when no API key can be resolved.
45    pub fn new(api_key: Option<&str>) -> Result<Self, TesseraError> {
46        Self::from_config(ClientConfig::new(api_key)?)
47    }
48
49    /// Create a client from a fully-specified [`ClientConfig`].
50    ///
51    /// # Errors
52    ///
53    /// Returns [`TesseraError::Network`] when the HTTP client or tokio runtime
54    /// cannot be built.
55    pub fn from_config(config: ClientConfig) -> Result<Self, TesseraError> {
56        let http = reqwest::blocking::Client::builder()
57            .redirect(reqwest::redirect::Policy::none())
58            .default_headers(config.auth_headers())
59            .timeout(config.timeout)
60            .build()
61            .map_err(|err| TesseraError::Network(err.to_string()))?;
62        #[cfg(feature = "polars")]
63        let runtime = tokio::runtime::Builder::new_multi_thread()
64            .enable_all()
65            .build()
66            .map_err(|err| TesseraError::Network(err.to_string()))?;
67        Ok(Self {
68            config,
69            http,
70            #[cfg(feature = "polars")]
71            runtime,
72        })
73    }
74
75    /// The resolved configuration this client was built with.
76    #[must_use]
77    pub fn config(&self) -> &ClientConfig {
78        &self.config
79    }
80
81    /// Close the underlying HTTP connection pool.
82    pub fn close(&mut self) {
83        // reqwest clients tear their pools down on drop; nothing else to do.
84    }
85
86    /// Send a prepared request, retrying transient failures.
87    fn request(
88        &self,
89        prepared: &PreparedRequest,
90    ) -> Result<reqwest::blocking::Response, TesseraError> {
91        for attempt in 0..=self.config.max_retries {
92            let last = attempt == self.config.max_retries;
93            let url = format!("{}{}", self.config.base_url, prepared.path);
94            let sent = self.http.get(url).query(&prepared.params).send();
95            match sent {
96                Err(err) if err.is_connect() || err.is_timeout() || err.is_request() => {
97                    if last {
98                        return Err(TesseraError::Network(format!(
99                            "network error contacting Tessera: {err}"
100                        )));
101                    }
102                    std::thread::sleep(backoff_delay(attempt, None));
103                }
104                Ok(response) if !last && should_retry(response.status().as_u16()) => {
105                    let retry_after = parse_retry_after(response.headers());
106                    std::thread::sleep(backoff_delay(attempt, retry_after));
107                }
108                Ok(response) => {
109                    if response.status().is_success() {
110                        return Ok(response);
111                    }
112                    let status = response.status().as_u16();
113                    let body = response
114                        .bytes()
115                        .map_err(|err| TesseraError::Network(err.to_string()))?;
116                    return Err(error_from_response(status, &body));
117                }
118                Err(err) => {
119                    return Err(TesseraError::Network(format!(
120                        "network error contacting Tessera: {err}"
121                    )));
122                }
123            }
124        }
125        unreachable!("retry loop always returns")
126    }
127
128    /// Parse a success response body as JSON.
129    fn json<T: serde::de::DeserializeOwned>(
130        response: reqwest::blocking::Response,
131    ) -> Result<T, TesseraError> {
132        let body = response
133            .bytes()
134            .map_err(|err| TesseraError::Network(err.to_string()))?;
135        serde_json::from_slice(&body).map_err(|err| TesseraError::Network(err.to_string()))
136    }
137
138    /// List every dataset visible to your plan.
139    ///
140    /// # Errors
141    ///
142    /// Returns any [`TesseraError`] the API or transport raises.
143    pub fn datasets(&self) -> Result<DatasetsResponse, TesseraError> {
144        let response = self.request(&datasets_request())?;
145        Self::json(response)
146    }
147
148    /// List the partitions of `asset`, optionally filtered by coin/month.
149    ///
150    /// # Errors
151    ///
152    /// Returns any [`TesseraError`] the API or transport raises.
153    pub fn partitions(
154        &self,
155        asset: &str,
156        coin: Option<&str>,
157        month: Option<&str>,
158    ) -> Result<PartitionsResponse, TesseraError> {
159        let response = self.request(&partitions_request(asset, coin, month))?;
160        Self::json(response)
161    }
162
163    /// Mint a short-lived presigned download URL for one partition.
164    ///
165    /// # Errors
166    ///
167    /// Returns any [`TesseraError`] the API or transport raises.
168    pub fn download_url(
169        &self,
170        asset: &str,
171        coin: &str,
172        month: &str,
173    ) -> Result<DownloadResponse, TesseraError> {
174        let response = self.request(&download_request(asset, coin, month))?;
175        Self::json(response)
176    }
177
178    /// Expand `(asset, coins, months)` into concrete partition references.
179    ///
180    /// # Errors
181    ///
182    /// Returns [`TesseraError::InvalidArgument`] on empty/invalid arguments.
183    pub fn partition_refs(
184        &self,
185        asset: &str,
186        coin: impl IntoCoins,
187        month: impl IntoMonths,
188    ) -> Result<Vec<PartitionRef>, TesseraError> {
189        let _ = self;
190        expand_refs(asset, coin, month)
191    }
192
193    /// Resolve presigned URLs for every ref, concurrently and order-preserving.
194    #[cfg(any(feature = "polars", feature = "duckdb"))]
195    fn resolve(&self, refs: &[PartitionRef]) -> Result<Vec<ResolvedPartition>, TesseraError> {
196        resolve_sync(
197            |partition| {
198                self.download_url(&partition.asset, &partition.coin, &partition.month)
199                    .map(|d| d.url)
200            },
201            refs,
202        )
203    }
204
205    /// Lazily scan one or more partitions into a Polars `LazyFrame`.
206    ///
207    /// URLs are minted now but the data is read on `.collect()`. Because
208    /// presigned URLs expire (~15 min), collect promptly; for long-lived
209    /// graphs re-run `scan` to refresh.
210    ///
211    /// # Errors
212    ///
213    /// Returns any [`TesseraError`] the API or transport raises.
214    #[cfg(feature = "polars")]
215    pub fn scan(
216        &self,
217        asset: &str,
218        coin: impl IntoCoins,
219        month: impl IntoMonths,
220        columns: Option<&[&str]>,
221    ) -> Result<LazyFrame, TesseraError> {
222        let parts = self.resolve(&expand_refs(asset, coin, month)?)?;
223        crate::readers::polars::build_lazyframe(&parts, columns)
224    }
225
226    /// Eagerly read one or more partitions into a Polars `DataFrame`.
227    ///
228    /// # Errors
229    ///
230    /// Returns any [`TesseraError`] the API or transport raises, including
231    /// [`TesseraError::PresignExpired`] for rejected presigned URLs.
232    #[cfg(feature = "polars")]
233    pub fn read(
234        &self,
235        asset: &str,
236        coin: impl IntoCoins,
237        month: impl IntoMonths,
238        columns: Option<&[&str]>,
239    ) -> Result<DataFrame, TesseraError> {
240        let parts = self.resolve(&expand_refs(asset, coin, month)?)?;
241        let lazy = crate::readers::polars::build_lazyframe(&parts, columns)?;
242        self.runtime
243            .block_on(async { crate::readers::polars::collect(lazy) })
244    }
245
246    /// Open one or more partitions as an in-memory DuckDB connection exposing
247    /// a `tessera` view for SQL querying.
248    ///
249    /// # Errors
250    ///
251    /// Returns any [`TesseraError`] the API or transport raises.
252    #[cfg(feature = "duckdb")]
253    pub fn to_duckdb(
254        &self,
255        asset: &str,
256        coin: impl IntoCoins,
257        month: impl IntoMonths,
258        columns: Option<&[&str]>,
259    ) -> Result<duckdb::Connection, TesseraError> {
260        let parts = self.resolve(&expand_refs(asset, coin, month)?)?;
261        crate::readers::duckdb::build_relation(&parts, columns)
262    }
263}
264
265impl Drop for TesseraClient {
266    fn drop(&mut self) {
267        self.close();
268    }
269}
270
271impl std::fmt::Debug for TesseraClient {
272    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
273        // Never print the API key.
274        f.debug_struct("TesseraClient")
275            .field("base_url", &self.config.base_url)
276            .field("timeout", &self.config.timeout)
277            .field("max_retries", &self.config.max_retries)
278            .finish_non_exhaustive()
279    }
280}
tessera/client.rs

tessera/
client.rs
