scirs2_io/database/
mod.rs

1//! Database connectivity for scientific data
2//!
3//! Provides interfaces for reading and writing scientific data to various
4//! database systems, including SQL and NoSQL databases.
5
6#![allow(dead_code)]
7#![allow(missing_docs)]
8#![allow(clippy::too_many_arguments)]
9
10use crate::error::{IoError, Result};
11use crate::metadata::Metadata;
12use scirs2_core::ndarray::{Array1, Array2, ArrayView2};
13use serde::{Deserialize, Serialize};
14use std::collections::HashMap;
15
16// Re-export database implementations
17#[cfg(feature = "sqlite")]
18pub mod sqlite;
19
20#[cfg(feature = "postgres")]
21pub mod postgres;
22
23#[cfg(feature = "mysql")]
24pub mod mysql;
25
26// Connection pooling
27pub mod pool;
28
29// Specialized modules
30pub mod bulk;
31pub mod table;
32pub mod timeseries;
33
34// Re-export commonly used types
35pub use self::pool::ConnectionPool;
36
37/// Supported database types
38#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
39pub enum DatabaseType {
40    /// PostgreSQL database
41    PostgreSQL,
42    /// MySQL/MariaDB database
43    MySQL,
44    /// SQLite database
45    SQLite,
46    /// MongoDB (NoSQL)
47    MongoDB,
48    /// InfluxDB (Time series)
49    InfluxDB,
50    /// Redis (Key-value)
51    Redis,
52    /// Cassandra (Wide column)
53    Cassandra,
54}
55
56/// Database connection configuration
57#[derive(Debug, Clone, Serialize, Deserialize)]
58pub struct DatabaseConfig {
59    /// The type of database (SQLite, PostgreSQL, etc.)
60    pub db_type: DatabaseType,
61    /// Host address for remote databases
62    pub host: Option<String>,
63    /// Port number for database connection
64    pub port: Option<u16>,
65    /// Database name or file path
66    pub database: String,
67    /// Username for authentication
68    pub username: Option<String>,
69    /// Password for authentication
70    pub password: Option<String>,
71    /// Additional connection options
72    pub options: HashMap<String, String>,
73}
74
75impl DatabaseConfig {
76    /// Create a new database configuration
77    pub fn new(db_type: DatabaseType, database: impl Into<String>) -> Self {
78        Self {
79            db_type,
80            host: None,
81            port: None,
82            database: database.into(),
83            username: None,
84            password: None,
85            options: HashMap::new(),
86        }
87    }
88
89    /// Set host
90    pub fn host(mut self, host: impl Into<String>) -> Self {
91        self.host = Some(host.into());
92        self
93    }
94
95    /// Set port
96    pub fn port(mut self, port: u16) -> Self {
97        self.port = Some(port);
98        self
99    }
100
101    /// Set credentials
102    pub fn credentials(mut self, username: impl Into<String>, password: impl Into<String>) -> Self {
103        self.username = Some(username.into());
104        self.password = Some(password.into());
105        self
106    }
107
108    /// Add connection option
109    pub fn option(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
110        self.options.insert(key.into(), value.into());
111        self
112    }
113
114    /// Build connection string
115    pub fn connection_string(&self) -> String {
116        match self.db_type {
117            DatabaseType::PostgreSQL => {
118                let host = self.host.as_deref().unwrap_or("localhost");
119                let port = self.port.unwrap_or(5432);
120                let user = self.username.as_deref().unwrap_or("postgres");
121                format!(
122                    "postgresql://{}:password@{}:{}/{}",
123                    user, host, port, self.database
124                )
125            }
126            DatabaseType::MySQL => {
127                let host = self.host.as_deref().unwrap_or("localhost");
128                let port = self.port.unwrap_or(3306);
129                let user = self.username.as_deref().unwrap_or("root");
130                format!(
131                    "mysql://{}:password@{}:{}/{}",
132                    user, host, port, self.database
133                )
134            }
135            DatabaseType::SQLite => {
136                format!("sqlite://{}", self.database)
137            }
138            DatabaseType::MongoDB => {
139                let host = self.host.as_deref().unwrap_or("localhost");
140                let port = self.port.unwrap_or(27017);
141                format!("mongodb://{}:{}/{}", host, port, self.database)
142            }
143            _ => format!("{}://{}", self.db_type.as_str(), self.database),
144        }
145    }
146}
147
148impl DatabaseType {
149    fn as_str(&self) -> &'static str {
150        match self {
151            Self::PostgreSQL => "postgresql",
152            Self::MySQL => "mysql",
153            Self::SQLite => "sqlite",
154            Self::MongoDB => "mongodb",
155            Self::InfluxDB => "influxdb",
156            Self::Redis => "redis",
157            Self::Cassandra => "cassandra",
158        }
159    }
160}
161
162/// Database query builder
163pub struct QueryBuilder {
164    pub(crate) query_type: QueryType,
165    pub(crate) table: String,
166    pub(crate) columns: Vec<String>,
167    pub(crate) conditions: Vec<String>,
168    pub(crate) values: Vec<serde_json::Value>,
169    pub(crate) order_by: Option<String>,
170    pub(crate) limit: Option<usize>,
171    pub(crate) offset: Option<usize>,
172}
173
174#[derive(Debug, Clone)]
175#[allow(dead_code)]
176pub(crate) enum QueryType {
177    Select,
178    Insert,
179    Update,
180    Delete,
181    CreateTable,
182}
183
184impl QueryBuilder {
185    /// Create a SELECT query
186    pub fn select(table: impl Into<String>) -> Self {
187        Self {
188            query_type: QueryType::Select,
189            table: table.into(),
190            columns: vec!["*".to_string()],
191            conditions: Vec::new(),
192            values: Vec::new(),
193            order_by: None,
194            limit: None,
195            offset: None,
196        }
197    }
198
199    /// Create an INSERT query
200    pub fn insert(table: impl Into<String>) -> Self {
201        Self {
202            query_type: QueryType::Insert,
203            table: table.into(),
204            columns: Vec::new(),
205            conditions: Vec::new(),
206            values: Vec::new(),
207            order_by: None,
208            limit: None,
209            offset: None,
210        }
211    }
212
213    /// Specify columns
214    pub fn columns(mut self, columns: Vec<impl Into<String>>) -> Self {
215        self.columns = columns.into_iter().map(|c| c.into()).collect();
216        self
217    }
218
219    /// Add WHERE condition
220    pub fn where_clause(mut self, condition: impl Into<String>) -> Self {
221        self.conditions.push(condition.into());
222        self
223    }
224
225    /// Add values for INSERT
226    pub fn values(mut self, values: Vec<serde_json::Value>) -> Self {
227        self.values = values;
228        self
229    }
230
231    /// Set ORDER BY
232    pub fn order_by(mut self, column: impl Into<String>, desc: bool) -> Self {
233        self.order_by = Some(format!(
234            "{} {}",
235            column.into(),
236            if desc { "DESC" } else { "ASC" }
237        ));
238        self
239    }
240
241    /// Set LIMIT
242    pub fn limit(mut self, limit: usize) -> Self {
243        self.limit = Some(limit);
244        self
245    }
246
247    /// Set OFFSET
248    pub fn offset(mut self, offset: usize) -> Self {
249        self.offset = Some(offset);
250        self
251    }
252
253    /// Build SQL query string
254    pub fn build_sql(&self) -> String {
255        match self.query_type {
256            QueryType::Select => {
257                let mut sql = format!("SELECT {} FROM {}", self.columns.join(", "), self.table);
258
259                if !self.conditions.is_empty() {
260                    sql.push_str(&format!(" WHERE {}", self.conditions.join(" AND ")));
261                }
262
263                if let Some(order) = &self.order_by {
264                    sql.push_str(&format!(" ORDER BY {order}"));
265                }
266
267                if let Some(limit) = self.limit {
268                    sql.push_str(&format!(" LIMIT {limit}"));
269                }
270
271                if let Some(offset) = self.offset {
272                    sql.push_str(&format!(" OFFSET {offset}"));
273                }
274
275                sql
276            }
277            QueryType::Insert => {
278                format!(
279                    "INSERT INTO {} ({}) VALUES ({})",
280                    self.table,
281                    self.columns.join(", "),
282                    self.values
283                        .iter()
284                        .map(|_| "?")
285                        .collect::<Vec<_>>()
286                        .join(", ")
287                )
288            }
289            _ => String::new(),
290        }
291    }
292
293    /// Build MongoDB query
294    pub fn build_mongo(&self) -> serde_json::Value {
295        match self.query_type {
296            QueryType::Select => {
297                let mut query = serde_json::json!({});
298
299                // Convert SQL-like conditions to MongoDB query
300                for condition in &self.conditions {
301                    // Simple parsing - in real implementation would be more sophisticated
302                    if let Some((field, value)) = condition.split_once(" = ") {
303                        query[field] = serde_json::json!(value.trim_matches('\''));
304                    }
305                }
306
307                serde_json::json!({
308                    "collection": self.table,
309                    "filter": query,
310                    "limit": self.limit,
311                    "skip": self.offset,
312                })
313            }
314            _ => serde_json::json!({}),
315        }
316    }
317}
318
319/// Database result set
320#[derive(Debug, Clone)]
321pub struct ResultSet {
322    /// Column names in the result set
323    pub columns: Vec<String>,
324    /// Data rows as JSON values
325    pub rows: Vec<Vec<serde_json::Value>>,
326    /// Additional metadata about the result set
327    pub metadata: Metadata,
328}
329
330impl ResultSet {
331    /// Create new result set
332    pub fn new(columns: Vec<String>) -> Self {
333        Self {
334            columns,
335            rows: Vec::new(),
336            metadata: Metadata::new(),
337        }
338    }
339
340    /// Add a row
341    pub fn add_row(&mut self, row: Vec<serde_json::Value>) {
342        self.rows.push(row);
343    }
344
345    /// Get number of rows
346    pub fn row_count(&self) -> usize {
347        self.rows.len()
348    }
349
350    /// Get number of columns
351    pub fn column_count(&self) -> usize {
352        self.columns.len()
353    }
354
355    /// Convert to `Array2<f64>` if all values are numeric
356    pub fn to_array(&self) -> Result<Array2<f64>> {
357        let mut data = Vec::new();
358
359        for row in &self.rows {
360            for value in row {
361                let num = value.as_f64().ok_or_else(|| {
362                    IoError::ConversionError("Non-numeric value in result set".to_string())
363                })?;
364                data.push(num);
365            }
366        }
367
368        Array2::from_shape_vec((self.row_count(), self.column_count()), data)
369            .map_err(|e| IoError::Other(e.to_string()))
370    }
371
372    /// Get column by name as Array1
373    pub fn get_column(&self, name: &str) -> Result<Array1<f64>> {
374        let col_idx = self
375            .columns
376            .iter()
377            .position(|c| c == name)
378            .ok_or_else(|| IoError::Other(format!("Column '{name}' not found")))?;
379
380        let mut data = Vec::new();
381        for row in &self.rows {
382            let num = row[col_idx].as_f64().ok_or_else(|| {
383                IoError::ConversionError("Non-numeric value in column".to_string())
384            })?;
385            data.push(num);
386        }
387
388        Ok(Array1::from_vec(data))
389    }
390}
391
392/// Database connection trait
393pub trait DatabaseConnection: Send + Sync {
394    /// Execute a query and return results
395    fn query(&self, query: &QueryBuilder) -> Result<ResultSet>;
396
397    /// Execute a raw SQL query
398    fn execute_sql(&self, sql: &str, params: &[serde_json::Value]) -> Result<ResultSet>;
399
400    /// Insert data from Array2
401    fn insert_array(&self, table: &str, data: ArrayView2<f64>, columns: &[&str]) -> Result<usize>;
402
403    /// Create table from schema
404    fn create_table(&self, table: &str, schema: &TableSchema) -> Result<()>;
405
406    /// Check if table exists
407    fn table_exists(&self, table: &str) -> Result<bool>;
408
409    /// Get table schema
410    fn get_schema(&self, table: &str) -> Result<TableSchema>;
411}
412
413/// Table schema definition
414#[derive(Debug, Clone, Serialize, Deserialize)]
415pub struct TableSchema {
416    /// Table name
417    pub name: String,
418    /// Column definitions
419    pub columns: Vec<ColumnDef>,
420    /// Primary key column names
421    pub primary_key: Option<Vec<String>>,
422    /// Index definitions
423    pub indexes: Vec<Index>,
424}
425
426/// Column definition for database tables
427#[derive(Debug, Clone, Serialize, Deserialize)]
428pub struct ColumnDef {
429    /// Column name
430    pub name: String,
431    /// Data type of the column
432    pub data_type: DataType,
433    /// Whether the column allows NULL values
434    pub nullable: bool,
435    /// Default value for the column
436    pub default: Option<serde_json::Value>,
437}
438
439/// Database data types
440#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
441#[serde(rename_all = "lowercase")]
442pub enum DataType {
443    /// 32-bit integer
444    Integer,
445    /// 64-bit integer
446    BigInt,
447    /// 32-bit floating point
448    Float,
449    /// 64-bit floating point
450    Double,
451    /// Decimal with precision and scale
452    Decimal(u8, u8),
453    /// Variable-length character string
454    Varchar(usize),
455    /// Text string of unlimited length
456    Text,
457    /// Boolean true/false
458    Boolean,
459    /// Date value
460    Date,
461    /// Timestamp with date and time
462    Timestamp,
463    /// JSON document
464    Json,
465    Binary,
466}
467
468#[derive(Debug, Clone, Serialize, Deserialize)]
469pub struct Index {
470    pub name: String,
471    pub columns: Vec<String>,
472    pub unique: bool,
473}
474
475/// Database connector factory
476pub struct DatabaseConnector;
477
478impl DatabaseConnector {
479    /// Create a new database connection
480    pub fn connect(config: &DatabaseConfig) -> Result<Box<dyn DatabaseConnection>> {
481        match config.db_type {
482            #[cfg(feature = "sqlite")]
483            DatabaseType::SQLite => Ok(Box::new(sqlite::SQLiteConnection::new(config)?)),
484            #[cfg(not(feature = "sqlite"))]
485            DatabaseType::SQLite => Err(IoError::UnsupportedFormat(
486                "SQLite support not enabled. Enable 'sqlite' feature.".to_string(),
487            )),
488
489            #[cfg(feature = "postgres")]
490            DatabaseType::PostgreSQL => Ok(Box::new(postgres::PostgreSQLConnection::new(config)?)),
491            #[cfg(not(feature = "postgres"))]
492            DatabaseType::PostgreSQL => Err(IoError::UnsupportedFormat(
493                "PostgreSQL support not enabled. Enable 'postgres' feature.".to_string(),
494            )),
495
496            #[cfg(feature = "mysql")]
497            DatabaseType::MySQL => Ok(Box::new(mysql::MySQLConnection::new(config)?)),
498            #[cfg(not(feature = "mysql"))]
499            DatabaseType::MySQL => Err(IoError::UnsupportedFormat(
500                "MySQL support not enabled. Enable 'mysql' feature.".to_string(),
501            )),
502
503            DatabaseType::MongoDB => Err(IoError::UnsupportedFormat(
504                "MongoDB support is not available in this build. \
505                 Use a dedicated MongoDB driver crate."
506                    .to_string(),
507            )),
508            DatabaseType::InfluxDB => Err(IoError::UnsupportedFormat(
509                "InfluxDB support is not available in this build. \
510                 Use a dedicated InfluxDB driver crate."
511                    .to_string(),
512            )),
513            DatabaseType::Redis => Err(IoError::UnsupportedFormat(
514                "Redis support is not available in this build. \
515                 Use a dedicated Redis driver crate."
516                    .to_string(),
517            )),
518            DatabaseType::Cassandra => Err(IoError::UnsupportedFormat(
519                "Cassandra support is not available in this build. \
520                 Use a dedicated Cassandra driver crate."
521                    .to_string(),
522            )),
523        }
524    }
525}
526
527#[cfg(test)]
528mod tests {
529    use super::*;
530
531    #[test]
532    fn test_database_config() {
533        let config = DatabaseConfig::new(DatabaseType::SQLite, "test.db");
534        assert_eq!(config.db_type, DatabaseType::SQLite);
535        assert_eq!(config.database, "test.db");
536        assert_eq!(config.connection_string(), "sqlite://test.db");
537    }
538
539    #[test]
540    fn test_query_builder() {
541        let query = QueryBuilder::select("users")
542            .columns(vec!["id", "name", "email"])
543            .where_clause("age > 21")
544            .limit(10);
545
546        let sql = query.build_sql();
547        assert!(sql.contains("SELECT id, name, email FROM users"));
548        assert!(sql.contains("WHERE age > 21"));
549        assert!(sql.contains("LIMIT 10"));
550    }
551}
scirs2_io/database/mod.rs

scirs2_io/database/
mod.rs
