nu_command/stor/
insert.rs

1use crate::database::{MEMORY_DB, SQLiteDatabase, get_shared_mem_conn, values_to_sql};
2use nu_engine::command_prelude::*;
3use nu_protocol::shell_error::generic::GenericError;
4use rusqlite::{Connection, params_from_iter};
5
6#[derive(Clone)]
7pub struct StorInsert;
8
9impl Command for StorInsert {
10    fn name(&self) -> &str {
11        "stor insert"
12    }
13
14    fn signature(&self) -> Signature {
15        Signature::build("stor insert")
16            .input_output_types(vec![
17                (Type::Nothing, Type::table()),
18                (Type::record(), Type::table()),
19                (Type::table(), Type::table()),
20                // FIXME Type::Any input added to disable pipeline input type checking, as run-time checks can raise undesirable type errors
21                // which aren't caught by the parser. see https://github.com/nushell/nushell/pull/14922 for more details
22                (Type::Any, Type::table()),
23            ])
24            .required_named(
25                "table-name",
26                SyntaxShape::String,
27                "Name of the table you want to insert into.",
28                Some('t'),
29            )
30            .named(
31                "data-record",
32                SyntaxShape::record(),
33                "A record of column names and column values to insert into the specified table.",
34                Some('d'),
35            )
36            .allow_variants_without_examples(true)
37            .category(Category::Database)
38    }
39
40    fn description(&self) -> &str {
41        "Insert information into a specified table in the in-memory sqlite database."
42    }
43
44    fn search_terms(&self) -> Vec<&str> {
45        vec!["sqlite", "storing", "table", "saving"]
46    }
47
48    fn examples(&self) -> Vec<Example<'_>> {
49        vec![
50            Example {
51                description: "Insert data in the in-memory sqlite database using a data-record of column-name and column-value pairs",
52                example: "stor insert --table-name nudb --data-record {bool1: true, int1: 5, float1: 1.1, str1: fdncred, datetime1: 2023-04-17}",
53                result: None,
54            },
55            Example {
56                description: "Insert data through pipeline input as a record of column-name and column-value pairs",
57                example: "{bool1: true, int1: 5, float1: 1.1, str1: fdncred, datetime1: 2023-04-17} | stor insert --table-name nudb",
58                result: None,
59            },
60            Example {
61                description: "Insert data through pipeline input as a table literal",
62                example: "[[bool1 int1 float1]; [true 5 1.1], [false 8 3.14]] | stor insert --table-name nudb",
63                result: None,
64            },
65            Example {
66                description: "Insert ls entries",
67                example: "ls | stor insert --table-name files",
68                result: None,
69            },
70            Example {
71                description: "Insert nu records as json data",
72                example: "ls -l | each {{file: $in.name, metadata: ($in | reject name)}} | stor insert --table-name files_with_md",
73                result: None,
74            },
75        ]
76    }
77
78    fn run(
79        &self,
80        engine_state: &EngineState,
81        stack: &mut Stack,
82        call: &Call,
83        input: PipelineData,
84    ) -> Result<PipelineData, ShellError> {
85        let span = call.head;
86        let table_name: Option<String> = call.get_flag(engine_state, stack, "table-name")?;
87        let data_record: Option<Record> = call.get_flag(engine_state, stack, "data-record")?;
88
89        let records = handle(span, data_record, input)?;
90
91        let conn = get_shared_mem_conn()?;
92        for record in records {
93            process(engine_state, table_name.clone(), span, &conn, record)?;
94        }
95
96        let db = Box::new(SQLiteDatabase::new(
97            std::path::Path::new(MEMORY_DB),
98            engine_state.signals().clone(),
99        ));
100        Ok(Value::custom(db, span).into_pipeline_data())
101    }
102}
103
104fn handle(
105    span: Span,
106    data_record: Option<Record>,
107    input: PipelineData,
108) -> Result<Vec<Record>, ShellError> {
109    // Check for conflicting use of both pipeline input and flag
110    if let Some(record) = data_record {
111        if !matches!(input, PipelineData::Empty) {
112            return Err(ShellError::Generic(GenericError::new(
113                "Pipeline and Flag both being used",
114                "Use either pipeline input or '--data-record' parameter",
115                span,
116            )));
117        }
118        return Ok(vec![record]);
119    }
120
121    // Handle the input types
122    let values = match input {
123        PipelineData::Empty => {
124            return Err(ShellError::MissingParameter {
125                param_name: "requires a table or a record".into(),
126                span,
127            });
128        }
129        PipelineData::ListStream(stream, ..) => stream.into_iter().collect::<Vec<_>>(),
130        PipelineData::Value(Value::List { vals, .. }, ..) => vals.into_owned(),
131        PipelineData::Value(val, ..) => vec![val],
132        _ => {
133            return Err(ShellError::OnlySupportsThisInputType {
134                exp_input_type: "list or record".into(),
135                wrong_type: "".into(),
136                dst_span: span,
137                src_span: span,
138            });
139        }
140    };
141
142    values
143        .into_iter()
144        .map(|val| match val {
145            Value::Record { val, .. } => Ok(val.into_owned()),
146            other => Err(ShellError::OnlySupportsThisInputType {
147                exp_input_type: "record".into(),
148                wrong_type: other.get_type().to_string(),
149                dst_span: span,
150                src_span: other.span(),
151            }),
152        })
153        .collect()
154}
155
156fn process(
157    engine_state: &EngineState,
158    table_name: Option<String>,
159    span: Span,
160    conn: &Connection,
161    record: Record,
162) -> Result<(), ShellError> {
163    if table_name.is_none() {
164        return Err(ShellError::MissingParameter {
165            param_name: "requires at table name".into(),
166            span,
167        });
168    }
169    let new_table_name = table_name.unwrap_or("table".into());
170
171    let mut create_stmt = format!("INSERT INTO {new_table_name} (");
172    let mut column_placeholders: Vec<String> = Vec::new();
173
174    let cols = record.columns();
175    cols.for_each(|col| {
176        column_placeholders.push(col.to_string());
177    });
178
179    create_stmt.push_str(&column_placeholders.join(", "));
180
181    // Values are set as placeholders.
182    create_stmt.push_str(") VALUES (");
183    let mut value_placeholders: Vec<String> = Vec::new();
184    for (index, _) in record.columns().enumerate() {
185        value_placeholders.push(format!("?{}", index + 1));
186    }
187    create_stmt.push_str(&value_placeholders.join(", "));
188    create_stmt.push(')');
189
190    // dbg!(&create_stmt);
191
192    // Get the params from the passed values
193    let params = values_to_sql(engine_state, record.values().cloned(), span)?;
194
195    conn.execute(&create_stmt, params_from_iter(params))
196        .map_err(|err| {
197            ShellError::Generic(GenericError::new_internal(
198                "Failed to insert using the SQLite connection to the in-memory database from insert.rs.",
199                err.to_string(),
200            ))
201        })?;
202    Ok(())
203}
204
205#[cfg(test)]
206mod test {
207    use chrono::DateTime;
208
209    use super::*;
210
211    #[test]
212    fn test_examples() -> nu_test_support::Result {
213        nu_test_support::test().examples(StorInsert)
214    }
215
216    #[test]
217    fn test_process_with_simple_parameters() {
218        let conn = get_shared_mem_conn().expect("Test was unable to get shared connection.");
219        let create_stmt = "CREATE TABLE test_process_with_simple_parameters (
220            int_column INTEGER,
221            real_column REAL,
222            str_column VARCHAR(255),
223            bool_column BOOLEAN,
224            date_column DATETIME DEFAULT(STRFTIME('%Y-%m-%d %H:%M:%f', 'NOW'))
225        )";
226
227        conn.execute(create_stmt, [])
228            .expect("Failed to create table as part of test.");
229        let table_name = Some("test_process_with_simple_parameters".to_string());
230        let span = Span::test_data();
231        let mut columns = Record::new();
232        columns.insert("int_column".to_string(), Value::test_int(42));
233        columns.insert("real_column".to_string(), Value::test_float(3.1));
234        columns.insert(
235            "str_column".to_string(),
236            Value::test_string("SimpleString".to_string()),
237        );
238        columns.insert("bool_column".to_string(), Value::test_bool(true));
239        columns.insert(
240            "date_column".to_string(),
241            Value::test_date(
242                DateTime::parse_from_str("2021-12-30 00:00:00 +0000", "%Y-%m-%d %H:%M:%S %z")
243                    .expect("Date string should parse."),
244            ),
245        );
246
247        let result = process(&EngineState::new(), table_name, span, &conn, columns);
248
249        assert!(result.is_ok());
250    }
251
252    #[test]
253    fn test_process_string_with_space() {
254        let conn = get_shared_mem_conn().expect("Test was unable to get shared connection.");
255        let create_stmt = "CREATE TABLE test_process_string_with_space (
256            str_column VARCHAR(255)
257        )";
258
259        conn.execute(create_stmt, [])
260            .expect("Failed to create table as part of test.");
261        let table_name = Some("test_process_string_with_space".to_string());
262        let span = Span::test_data();
263        let mut columns = Record::new();
264        columns.insert(
265            "str_column".to_string(),
266            Value::test_string("String With Spaces".to_string()),
267        );
268
269        let result = process(&EngineState::new(), table_name, span, &conn, columns);
270
271        assert!(result.is_ok());
272    }
273
274    #[test]
275    fn test_no_errors_when_string_too_long() {
276        let conn = get_shared_mem_conn().expect("Test was unable to get shared connection.");
277        let create_stmt = "CREATE TABLE test_errors_when_string_too_long (
278            str_column VARCHAR(8)
279        )";
280
281        conn.execute(create_stmt, [])
282            .expect("Failed to create table as part of test.");
283        let table_name = Some("test_errors_when_string_too_long".to_string());
284        let span = Span::test_data();
285        let mut columns = Record::new();
286        columns.insert(
287            "str_column".to_string(),
288            Value::test_string("ThisIsALongString".to_string()),
289        );
290
291        let result = process(&EngineState::new(), table_name, span, &conn, columns);
292        // SQLite uses dynamic typing, making any length acceptable for a varchar column
293        assert!(result.is_ok());
294    }
295
296    #[test]
297    fn test_no_errors_when_param_is_wrong_type() {
298        let conn = get_shared_mem_conn().expect("Test was unable to get shared connection.");
299        let create_stmt = "CREATE TABLE test_errors_when_param_is_wrong_type (
300            int_column INT
301        )";
302
303        conn.execute(create_stmt, [])
304            .expect("Failed to create table as part of test.");
305        let table_name = Some("test_errors_when_param_is_wrong_type".to_string());
306        let span = Span::test_data();
307        let mut columns = Record::new();
308        columns.insert(
309            "int_column".to_string(),
310            Value::test_string("ThisIsTheWrongType".to_string()),
311        );
312
313        let result = process(&EngineState::new(), table_name, span, &conn, columns);
314        // SQLite uses dynamic typing, making any type acceptable for a column
315        assert!(result.is_ok());
316    }
317
318    #[test]
319    fn test_errors_when_column_doesnt_exist() {
320        let conn = get_shared_mem_conn().expect("Test was unable to get shared connection.");
321        let create_stmt = "CREATE TABLE test_errors_when_column_doesnt_exist (
322            int_column INT
323        )";
324
325        conn.execute(create_stmt, [])
326            .expect("Failed to create table as part of test.");
327        let table_name = Some("test_errors_when_column_doesnt_exist".to_string());
328        let span = Span::test_data();
329        let mut columns = Record::new();
330        columns.insert(
331            "not_a_column".to_string(),
332            Value::test_string("ThisIsALongString".to_string()),
333        );
334
335        let result = process(&EngineState::new(), table_name, span, &conn, columns);
336
337        assert!(result.is_err());
338    }
339
340    #[test]
341    fn test_errors_when_table_doesnt_exist() {
342        let conn = get_shared_mem_conn().expect("Test was unable to get shared connection.");
343
344        let table_name = Some("test_errors_when_table_doesnt_exist".to_string());
345        let span = Span::test_data();
346        let mut columns = Record::new();
347        columns.insert(
348            "str_column".to_string(),
349            Value::test_string("ThisIsALongString".to_string()),
350        );
351
352        let result = process(&EngineState::new(), table_name, span, &conn, columns);
353
354        assert!(result.is_err());
355    }
356
357    #[test]
358    fn test_insert_json() {
359        let conn = get_shared_mem_conn().expect("Test was unable to get shared connection.");
360
361        let create_stmt = "CREATE TABLE test_insert_json (
362            json_field JSON,
363            jsonb_field JSONB 
364        )";
365
366        conn.execute(create_stmt, [])
367            .expect("Failed to create table as part of test.");
368
369        let mut record = Record::new();
370        record.insert("x", Value::test_int(89));
371        record.insert("y", Value::test_int(12));
372        record.insert(
373            "z",
374            Value::test_list(vec![
375                Value::test_string("hello"),
376                Value::test_string("goodbye"),
377            ]),
378        );
379
380        let mut row = Record::new();
381        row.insert("json_field", Value::test_record(record.clone()));
382        row.insert("jsonb_field", Value::test_record(record));
383
384        let result = process(
385            &EngineState::new(),
386            Some("test_insert_json".to_owned()),
387            Span::test_data(),
388            &conn,
389            row,
390        );
391
392        assert!(result.is_ok());
393    }
394}
nu_command/stor/insert.rs

nu_command/stor/
insert.rs
