faucet_source_postgres/
config.rs

1//! PostgreSQL source configuration.
2
3use schemars::JsonSchema;
4use serde::{Deserialize, Serialize};
5use serde_json::Value;
6
7/// Configuration for the PostgreSQL query source.
8#[derive(Clone, Serialize, Deserialize, JsonSchema)]
9pub struct PostgresSourceConfig {
10    /// PostgreSQL connection URL (e.g. `postgres://user:pass@host/db`).
11    pub connection_url: String,
12    /// SQL query to execute.
13    pub query: String,
14    /// Bind parameters for the query. Defaults to empty.
15    #[serde(default)]
16    pub params: Vec<Value>,
17    /// Maximum number of connections in the pool. Defaults to 10.
18    #[serde(default = "default_max_connections")]
19    pub max_connections: u32,
20}
21
22fn default_max_connections() -> u32 {
23    10
24}
25
26impl std::fmt::Debug for PostgresSourceConfig {
27    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
28        f.debug_struct("PostgresSourceConfig")
29            .field("connection_url", &"***")
30            .field("query", &self.query)
31            .field("params", &self.params)
32            .field("max_connections", &self.max_connections)
33            .finish()
34    }
35}
36
37impl PostgresSourceConfig {
38    /// Create a new config with the required connection URL and query.
39    pub fn new(connection_url: impl Into<String>, query: impl Into<String>) -> Self {
40        Self {
41            connection_url: connection_url.into(),
42            query: query.into(),
43            params: Vec::new(),
44            max_connections: 10,
45        }
46    }
47
48    /// Set bind parameters for the query.
49    pub fn params(mut self, params: Vec<Value>) -> Self {
50        self.params = params;
51        self
52    }
53
54    /// Set the maximum number of connections in the pool.
55    pub fn with_max_connections(mut self, max_connections: u32) -> Self {
56        self.max_connections = max_connections;
57        self
58    }
59}
60
61#[cfg(test)]
62mod tests {
63    use super::*;
64    use serde_json::json;
65
66    #[test]
67    fn default_config() {
68        let config = PostgresSourceConfig::new("postgres://localhost/test", "SELECT * FROM events");
69        assert_eq!(config.query, "SELECT * FROM events");
70        assert!(config.params.is_empty());
71    }
72
73    #[test]
74    fn builder_with_params() {
75        let config = PostgresSourceConfig::new(
76            "postgres://localhost/test",
77            "SELECT * FROM events WHERE id = $1",
78        )
79        .params(vec![json!(42)]);
80        assert_eq!(config.params.len(), 1);
81        assert_eq!(config.params[0], json!(42));
82    }
83
84    #[test]
85    fn debug_masks_connection_url() {
86        let config = PostgresSourceConfig::new("postgres://secret:pass@host/db", "SELECT 1");
87        let debug = format!("{config:?}");
88        assert!(debug.contains("***"));
89        assert!(!debug.contains("secret"));
90        assert!(!debug.contains("pass"));
91    }
92}
faucet_source_postgres/config.rs

faucet_source_postgres/
config.rs
