influxdb3_client/
config.rs

1use std::time::Duration;
2
3use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
4use url::Url;
5
6use crate::{error::Error, precision::Precision, retry::RetryConfig, write::WriteOptions};
7
8/// Configuration for the InfluxDB 3 client.
9///
10/// Construct with [`ClientConfig::builder()`] or parse from a connection string /
11/// environment variables with [`ClientConfig::from_connection_string()`] /
12/// [`ClientConfig::from_env()`].
13#[derive(Debug, Clone)]
14pub struct ClientConfig {
15    /// InfluxDB host URL (e.g. `https://cluster.influxdata.io`).
16    pub host: String,
17
18    /// API token.
19    pub token: Option<String>,
20
21    /// Authentication scheme: `"Bearer"` (default) or `"Token"`.
22    pub auth_scheme: String,
23
24    /// Database for all operations. Required; validated at construction time.
25    pub database: String,
26
27    /// Organization name (used for V2 API compatibility).
28    pub org: Option<String>,
29
30    /// Default write options applied to every write call.
31    pub write_options: WriteOptions,
32
33    /// Default retry policy for transient write/query failures. Override per
34    /// request with `WriteRequest`/`QueryRequest` `.retry()` / `.no_retry()`.
35    pub retry: RetryConfig,
36
37    /// Extra HTTP headers sent with every request.
38    pub headers: HeaderMap,
39
40    /// Path to a PEM file with additional CA roots for TLS verification.
41    pub ssl_roots_path: Option<String>,
42
43    /// HTTP proxy URL.
44    pub proxy: Option<String>,
45
46    /// Request timeout for write calls.
47    pub write_timeout: Duration,
48
49    /// Timeout for the Flight channel connect and for collected (`.await`)
50    /// queries. Streaming queries (`.stream()`) are intentionally unbounded.
51    pub query_timeout: Duration,
52
53    /// Keep-alive idle connection timeout.
54    pub idle_connection_timeout: Duration,
55
56    /// Maximum number of idle connections in the pool.
57    pub max_idle_connections: usize,
58}
59
60impl Default for ClientConfig {
61    fn default() -> Self {
62        ClientConfig {
63            host: String::new(),
64            token: None,
65            auth_scheme: "Bearer".to_string(),
66            database: String::new(), // validated as non-empty in build()
67            org: None,
68            write_options: WriteOptions::default(),
69            retry: RetryConfig::default(),
70            headers: HeaderMap::new(),
71            ssl_roots_path: None,
72            proxy: None,
73            write_timeout: Duration::from_secs(30),
74            query_timeout: Duration::from_secs(60),
75            idle_connection_timeout: Duration::from_secs(90),
76            max_idle_connections: 100,
77        }
78    }
79}
80
81impl ClientConfig {
82    /// Start building a config.
83    pub fn builder() -> ClientConfigBuilder {
84        ClientConfigBuilder::default()
85    }
86
87    /// Parse client configuration from process environment variables.
88    ///
89    /// Supported variables:
90    /// - `INFLUX_HOST` - InfluxDB host URL (required).
91    /// - `INFLUX_DATABASE` - database name (required).
92    /// - `INFLUX_TOKEN` - authentication token.
93    /// - `INFLUX_AUTH_SCHEME` - authentication scheme.
94    /// - `INFLUX_ORG` - organization name.
95    /// - `INFLUX_PRECISION` - write precision (`ns`, `us`, `ms`, `s`, or long form).
96    /// - `INFLUX_GZIP_THRESHOLD` - gzip threshold in bytes.
97    /// - `INFLUX_WRITE_NO_SYNC` - skip WAL synchronization for writes.
98    /// - `INFLUX_WRITE_ACCEPT_PARTIAL` - accept partial writes.
99    /// - `INFLUX_WRITE_USE_V2_API` - use the V2 write endpoint.
100    pub fn from_env() -> Result<Self, Error> {
101        let host = std::env::var("INFLUX_HOST").map_err(|_| Error::EnvVar("INFLUX_HOST".into()))?;
102        let database = std::env::var("INFLUX_DATABASE")
103            .map_err(|_| Error::EnvVar("INFLUX_DATABASE".into()))?;
104
105        let token = std::env::var("INFLUX_TOKEN").ok();
106        let auth_scheme = std::env::var("INFLUX_AUTH_SCHEME").ok();
107        let org = std::env::var("INFLUX_ORG").ok();
108        let mut write_options = WriteOptions::default();
109        if let Ok(value) = std::env::var("INFLUX_PRECISION") {
110            write_options.precision = parse_precision(&value)?;
111        }
112        if let Ok(value) = std::env::var("INFLUX_GZIP_THRESHOLD") {
113            write_options.gzip_threshold = Some(parse_usize("INFLUX_GZIP_THRESHOLD", &value)?);
114        }
115        if let Ok(value) = std::env::var("INFLUX_WRITE_NO_SYNC") {
116            write_options.no_sync = parse_bool("INFLUX_WRITE_NO_SYNC", &value)?;
117        }
118        if let Ok(value) = std::env::var("INFLUX_WRITE_ACCEPT_PARTIAL") {
119            write_options.accept_partial = parse_bool("INFLUX_WRITE_ACCEPT_PARTIAL", &value)?;
120        }
121        if let Ok(value) = std::env::var("INFLUX_WRITE_USE_V2_API") {
122            write_options.use_v2_api = parse_bool("INFLUX_WRITE_USE_V2_API", &value)?;
123        }
124
125        let mut builder = ClientConfig::builder()
126            .host(host)
127            .database(database)
128            .token_opt(token)
129            .org_opt(org)
130            .write_options(write_options);
131        if let Some(auth_scheme) = auth_scheme {
132            builder = builder.auth_scheme(auth_scheme);
133        }
134        builder.build()
135    }
136
137    /// Parse a URL-formatted connection string, e.g.:
138    ///
139    /// ```text
140    /// https://cluster.influxdata.io/?token=TOKEN&database=DB&org=ORG
141    /// ```
142    ///
143    /// Supported query parameters:
144    /// - `token` - authentication token.
145    /// - `database` - database name (required).
146    /// - `org` - organization name.
147    /// - `authScheme` - authentication scheme.
148    /// - `precision` - write precision (`ns`, `us`, `ms`, `s`, or long form).
149    /// - `gzipThreshold` - gzip threshold in bytes.
150    /// - `writeNoSync` - skip WAL synchronization for writes.
151    /// - `writeAcceptPartial` - accept partial writes.
152    /// - `writeUseV2Api` - use the V2 write endpoint.
153    pub fn from_connection_string(cs: &str) -> Result<Self, Error> {
154        let url = Url::parse(cs)?;
155        let mut host_url = url.clone();
156        host_url
157            .set_password(None)
158            .map_err(|_| Error::Config("invalid connection string host".into()))?;
159        host_url
160            .set_username("")
161            .map_err(|_| Error::Config("invalid connection string host".into()))?;
162        host_url.set_path("");
163        host_url.set_query(None);
164        host_url.set_fragment(None);
165        let host = host_url.to_string();
166
167        let mut builder = ClientConfig::builder().host(host);
168        let mut write_options = WriteOptions::default();
169
170        for (key, value) in url.query_pairs() {
171            match key.as_ref() {
172                "token" => {
173                    builder = builder.token(value.into_owned());
174                }
175                "database" => {
176                    builder = builder.database(value.into_owned());
177                }
178                "org" => {
179                    builder = builder.org(value.into_owned());
180                }
181                "authScheme" => {
182                    builder = builder.auth_scheme(value.into_owned());
183                }
184                "precision" => {
185                    write_options.precision = parse_precision(value.as_ref())?;
186                }
187                "gzipThreshold" => {
188                    write_options.gzip_threshold =
189                        Some(parse_usize("gzipThreshold", value.as_ref())?);
190                }
191                "writeNoSync" => {
192                    write_options.no_sync = parse_bool("writeNoSync", value.as_ref())?;
193                }
194                "writeAcceptPartial" => {
195                    write_options.accept_partial =
196                        parse_bool("writeAcceptPartial", value.as_ref())?;
197                }
198                "writeUseV2Api" => {
199                    write_options.use_v2_api = parse_bool("writeUseV2Api", value.as_ref())?;
200                }
201                _other => {}
202            }
203        }
204
205        builder.write_options(write_options).build()
206    }
207
208    /// Return the normalised host URL (trailing slash stripped).
209    pub fn host_url(&self) -> &str {
210        self.host.trim_end_matches('/')
211    }
212
213    /// Build the `Authorization` header value (`"Bearer TOKEN"` etc.).
214    ///
215    /// Returns `Ok(None)` when no token is set. Returns an error if the token
216    /// contains characters that are invalid in an HTTP header value.
217    pub fn authorization_header(&self) -> Result<Option<HeaderValue>, Error> {
218        match &self.token {
219            None => Ok(None),
220            Some(tok) => HeaderValue::from_str(&format!("{} {}", self.auth_scheme, tok))
221                .map(Some)
222                .map_err(|_| Error::Config("token contains invalid header characters".into())),
223        }
224    }
225}
226
227fn parse_precision(value: &str) -> Result<Precision, Error> {
228    value
229        .parse()
230        .map_err(|e| Error::Config(format!("invalid precision '{value}': {e}")))
231}
232
233fn parse_usize(name: &str, value: &str) -> Result<usize, Error> {
234    value
235        .parse()
236        .map_err(|e| Error::Config(format!("invalid {name} '{value}': {e}")))
237}
238
239fn parse_bool(name: &str, value: &str) -> Result<bool, Error> {
240    value
241        .parse()
242        .map_err(|e| Error::Config(format!("invalid {name} '{value}': {e}")))
243}
244
245/// Fluent builder for [`ClientConfig`].
246#[derive(Debug, Default)]
247pub struct ClientConfigBuilder {
248    cfg: ClientConfig,
249    /// Validated when [`ClientConfigBuilder::build`] is called, so a malformed
250    /// header surfaces as an error rather than a panic at insertion time.
251    pending_headers: Vec<(String, String)>,
252}
253
254impl ClientConfigBuilder {
255    /// Required: the InfluxDB host URL.
256    pub fn host(mut self, host: impl Into<String>) -> Self {
257        self.cfg.host = host.into();
258        self
259    }
260
261    pub fn token(mut self, token: impl Into<String>) -> Self {
262        self.cfg.token = Some(token.into());
263        self
264    }
265
266    pub fn token_opt(mut self, token: Option<String>) -> Self {
267        self.cfg.token = token;
268        self
269    }
270
271    /// `"Bearer"` (default) or `"Token"`.
272    pub fn auth_scheme(mut self, scheme: impl Into<String>) -> Self {
273        self.cfg.auth_scheme = scheme.into();
274        self
275    }
276
277    pub fn database(mut self, db: impl Into<String>) -> Self {
278        self.cfg.database = db.into();
279        self
280    }
281
282    pub fn org(mut self, org: impl Into<String>) -> Self {
283        self.cfg.org = Some(org.into());
284        self
285    }
286
287    pub fn org_opt(mut self, org: Option<String>) -> Self {
288        self.cfg.org = org;
289        self
290    }
291
292    pub fn write_options(mut self, opts: WriteOptions) -> Self {
293        self.cfg.write_options = opts;
294        self
295    }
296
297    /// Set whether writes use the V2 `/api/v2/write` endpoint by default.
298    pub fn write_use_v2_api(mut self, use_v2_api: bool) -> Self {
299        self.cfg.write_options.use_v2_api = use_v2_api;
300        self
301    }
302
303    /// Set whether V3 writes can partially succeed when some lines fail.
304    pub fn write_accept_partial(mut self, accept_partial: bool) -> Self {
305        self.cfg.write_options.accept_partial = accept_partial;
306        self
307    }
308
309    /// Set whether V3 writes skip WAL synchronization by default.
310    pub fn write_no_sync(mut self, no_sync: bool) -> Self {
311        self.cfg.write_options.no_sync = no_sync;
312        self
313    }
314
315    /// Set the default retry policy for transient write/query failures.
316    pub fn retry(mut self, retry: RetryConfig) -> Self {
317        self.cfg.retry = retry;
318        self
319    }
320
321    /// Add a single extra HTTP header sent with every request.
322    ///
323    /// The name and value are validated in [`build`](Self::build), so an
324    /// invalid header is reported as an error rather than panicking here.
325    pub fn header(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
326        self.pending_headers.push((key.into(), value.into()));
327        self
328    }
329
330    pub fn ssl_roots_path(mut self, path: impl Into<String>) -> Self {
331        self.cfg.ssl_roots_path = Some(path.into());
332        self
333    }
334
335    pub fn proxy(mut self, proxy: impl Into<String>) -> Self {
336        self.cfg.proxy = Some(proxy.into());
337        self
338    }
339
340    pub fn write_timeout(mut self, dur: Duration) -> Self {
341        self.cfg.write_timeout = dur;
342        self
343    }
344
345    pub fn query_timeout(mut self, dur: Duration) -> Self {
346        self.cfg.query_timeout = dur;
347        self
348    }
349
350    pub fn idle_connection_timeout(mut self, dur: Duration) -> Self {
351        self.cfg.idle_connection_timeout = dur;
352        self
353    }
354
355    pub fn max_idle_connections(mut self, n: usize) -> Self {
356        self.cfg.max_idle_connections = n;
357        self
358    }
359
360    /// Validate and produce the final [`ClientConfig`].
361    ///
362    /// Returns an error if `host` or `database` were not set.
363    pub fn build(mut self) -> Result<ClientConfig, Error> {
364        if self.cfg.host.is_empty() {
365            return Err(Error::Config("host is required".into()));
366        }
367        Url::parse(&self.cfg.host)
368            .map_err(|e| Error::Config(format!("invalid host URL '{}': {e}", self.cfg.host)))?;
369        if self.cfg.database.is_empty() {
370            return Err(Error::Config("database is required".into()));
371        }
372        self.cfg.write_options.validate()?;
373
374        for (key, value) in self.pending_headers {
375            let name = HeaderName::from_bytes(key.as_bytes())
376                .map_err(|e| Error::Config(format!("invalid header name '{key}': {e}")))?;
377            let val = HeaderValue::from_str(&value)
378                .map_err(|e| Error::Config(format!("invalid value for header '{key}': {e}")))?;
379            self.cfg.headers.insert(name, val);
380        }
381
382        // Surface a malformed token now rather than on the first request.
383        self.cfg.authorization_header()?;
384
385        Ok(self.cfg)
386    }
387}
influxdb3_client/config.rs

influxdb3_client/
config.rs
