Allow stor insert and stor update to accept pipeline input (#12882)

- this PR should close #11433 

# Description
This PR implements pipeline input support for the stor insert and stor
update commands,
enabling users to directly pass data to these commands without relying
solely on flag parameters.

Previously, it was only possible to specify the record data using flag
parameters,
which could be less intuitive and become cumbersome:

```bash
stor insert --table-name nudb --data-record {bool1: true, int1: 5, float1: 1.1, str1: fdncred, datetime1: 2023-04-17}
stor update --table-name nudb --update-record {str1: nushell datetime1: 2020-04-17}
```

Now it is also possible to pass a record through pipeline input:

```bash
{bool1: true, int1: 5, float1: 1.1, str1: fdncred, datetime1: 2023-04-17} | stor insert --table-name nudb
{str1: nushell datetime1: 2020-04-17} | stor update --table-name nudb"
```

Changes made on code:

- Modified stor insert and stor update to accept a record from the
pipeline.
- Added logic to handle data from the pipeline record.
- Implemented an error case to prevent simultaneous data input from both
pipeline and flag.

# User-facing changes

Returns an error when both ways of inserting data are used.

The examples for both commands were updated and in each command, when
the -d or -u fags are being used at the same time as input is being
passed through the pipeline, it returns an error:


![image](https://github.com/nushell/nushell/assets/120738170/c5b15c1b-716a-4df4-95e8-3bca8f7ae224)

Also returns an error when both of them are missing:


![image](https://github.com/nushell/nushell/assets/120738170/47f538ab-79f1-4fcc-9c62-d7a7d60f86a1)


# Tests + Formating
- 🟢 `toolkit fmt`
- 🟢 `toolkit clippy`
- 🟢 `toolkit test`
- 🟢 `toolkit test stdlib`

Co-authored-by: Rodrigo Friães <rodrigo.friaes@tecnico.ulisboa.pt>

crates/nu-command/src/stor/insert.rs

@@ -12,14 +12,17 @@ impl Command for StorInsert {
 
     fn signature(&self) -> Signature {
         Signature::build("stor insert")
-            .input_output_types(vec![(Type::Nothing, Type::table())])
+            .input_output_types(vec![
+                (Type::Nothing, Type::table()),
+                (Type::record(), Type::table()),
+            ])
             .required_named(
                 "table-name",
                 SyntaxShape::String,
                 "name of the table you want to insert into",
                 Some('t'),
             )
-            .required_named(
+            .named(
                 "data-record",
                 SyntaxShape::Record(vec![]),
                 "a record of column names and column values to insert into the specified table",
@@ -39,10 +42,16 @@ impl Command for StorInsert {
 
     fn examples(&self) -> Vec<Example> {
         vec![Example {
-            description: "Insert data the in-memory sqlite database using a data-record of column-name and column-value pairs",
-            example: "stor insert --table-name nudb --data-record {bool1: true, int1: 5, float1: 1.1, str1: fdncred, datetime1: 2023-04-17}",
-            result: None,
-        }]
+                description: "Insert data the in-memory sqlite database using a data-record of column-name and column-value pairs",
+                example: "stor insert --table-name nudb --data-record {bool1: true, int1: 5, float1: 1.1, str1: fdncred, datetime1: 2023-04-17}",
+                result: None,
+            },
+            Example {
+                description: "Insert data through pipeline input as a record of column-name and column-value pairs",
+                example: "{bool1: true, int1: 5, float1: 1.1, str1: fdncred, datetime1: 2023-04-17} | stor insert --table-name nudb",
+                result: None,
+            },
+        ]
     }
 
     fn run(
@@ -50,25 +59,79 @@ impl Command for StorInsert {
         engine_state: &EngineState,
         stack: &mut Stack,
         call: &Call,
-        _input: PipelineData,
+        input: PipelineData,
     ) -> Result<PipelineData, ShellError> {
         let span = call.head;
         let table_name: Option<String> = call.get_flag(engine_state, stack, "table-name")?;
-        let columns: Option<Record> = call.get_flag(engine_state, stack, "data-record")?;
+        let data_record: Option<Record> = call.get_flag(engine_state, stack, "data-record")?;
         // let config = engine_state.get_config();
         let db = Box::new(SQLiteDatabase::new(std::path::Path::new(MEMORY_DB), None));
 
+        // Check if the record is being passed as input or using the data record parameter
+        let columns = handle(span, data_record, input)?;
+
         process(table_name, span, &db, columns)?;
 
         Ok(Value::custom(db, span).into_pipeline_data())
     }
 }
 
+fn handle(
+    span: Span,
+    data_record: Option<Record>,
+    input: PipelineData,
+) -> Result<Record, ShellError> {
+    match input {
+        PipelineData::Empty => data_record.ok_or_else(|| ShellError::MissingParameter {
+            param_name: "requires a record".into(),
+            span,
+        }),
+        PipelineData::Value(value, ..) => {
+            // Since input is being used, check if the data record parameter is used too
+            if data_record.is_some() {
+                return Err(ShellError::GenericError {
+                    error: "Pipeline and Flag both being used".into(),
+                    msg: "Use either pipeline input or '--data-record' parameter".into(),
+                    span: Some(span),
+                    help: None,
+                    inner: vec![],
+                });
+            }
+            match value {
+                Value::Record { val, .. } => Ok(val.into_owned()),
+                val => Err(ShellError::OnlySupportsThisInputType {
+                    exp_input_type: "record".into(),
+                    wrong_type: val.get_type().to_string(),
+                    dst_span: Span::unknown(),
+                    src_span: val.span(),
+                }),
+            }
+        }
+        _ => {
+            if data_record.is_some() {
+                return Err(ShellError::GenericError {
+                    error: "Pipeline and Flag both being used".into(),
+                    msg: "Use either pipeline input or '--data-record' parameter".into(),
+                    span: Some(span),
+                    help: None,
+                    inner: vec![],
+                });
+            }
+            Err(ShellError::OnlySupportsThisInputType {
+                exp_input_type: "record".into(),
+                wrong_type: "".into(),
+                dst_span: span,
+                src_span: span,
+            })
+        }
+    }
+}
+
 fn process(
     table_name: Option<String>,
     span: Span,
     db: &SQLiteDatabase,
-    columns: Option<Record>,
+    record: Record,
 ) -> Result<(), ShellError> {
     if table_name.is_none() {
         return Err(ShellError::MissingParameter {
@@ -77,54 +140,45 @@ fn process(
         });
     }
     let new_table_name = table_name.unwrap_or("table".into());
+
     if let Ok(conn) = db.open_connection() {
-        match columns {
-            Some(record) => {
-                let mut create_stmt = format!("INSERT INTO {} ( ", new_table_name);
-                let cols = record.columns();
-                cols.for_each(|col| {
-                    create_stmt.push_str(&format!("{}, ", col));
-                });
-                if create_stmt.ends_with(", ") {
-                    create_stmt.pop();
-                    create_stmt.pop();
-                }
+        let mut create_stmt = format!("INSERT INTO {} ( ", new_table_name);
+        let cols = record.columns();
+        cols.for_each(|col| {
+            create_stmt.push_str(&format!("{}, ", col));
+        });
+        if create_stmt.ends_with(", ") {
+            create_stmt.pop();
+            create_stmt.pop();
+        }
 
-                // Values are set as placeholders.
-                create_stmt.push_str(") VALUES ( ");
-                for (index, _) in record.columns().enumerate() {
-                    create_stmt.push_str(&format!("?{}, ", index + 1));
-                }
+        // Values are set as placeholders.
+        create_stmt.push_str(") VALUES ( ");
+        for (index, _) in record.columns().enumerate() {
+            create_stmt.push_str(&format!("?{}, ", index + 1));
+        }
 
-                if create_stmt.ends_with(", ") {
-                    create_stmt.pop();
-                    create_stmt.pop();
-                }
+        if create_stmt.ends_with(", ") {
+            create_stmt.pop();
+            create_stmt.pop();
+        }
 
-                create_stmt.push(')');
+        create_stmt.push(')');
 
-                // dbg!(&create_stmt);
+        // dbg!(&create_stmt);
 
-                // Get the params from the passed values
-                let params = values_to_sql(record.values().cloned())?;
+        // Get the params from the passed values
+        let params = values_to_sql(record.values().cloned())?;
 
-                conn.execute(&create_stmt, params_from_iter(params))
-                    .map_err(|err| ShellError::GenericError {
-                        error: "Failed to open SQLite connection in memory from insert".into(),
-                        msg: err.to_string(),
-                        span: Some(Span::test_data()),
-                        help: None,
-                        inner: vec![],
-                    })?;
-            }
-            None => {
-                return Err(ShellError::MissingParameter {
-                    param_name: "requires at least one column".into(),
-                    span,
-                });
-            }
-        };
-    }
+        conn.execute(&create_stmt, params_from_iter(params))
+            .map_err(|err| ShellError::GenericError {
+                error: "Failed to open SQLite connection in memory from insert".into(),
+                msg: err.to_string(),
+                span: Some(Span::test_data()),
+                help: None,
+                inner: vec![],
+            })?;
+    };
     // dbg!(db.clone());
     Ok(())
 }
@@ -176,7 +230,7 @@ mod test {
             ),
         );
 
-        let result = process(table_name, span, &db, Some(columns));
+        let result = process(table_name, span, &db, columns);
 
         assert!(result.is_ok());
     }
@@ -201,7 +255,7 @@ mod test {
             Value::test_string("String With Spaces".to_string()),
         );
 
-        let result = process(table_name, span, &db, Some(columns));
+        let result = process(table_name, span, &db, columns);
 
         assert!(result.is_ok());
     }
@@ -226,7 +280,7 @@ mod test {
             Value::test_string("ThisIsALongString".to_string()),
         );
 
-        let result = process(table_name, span, &db, Some(columns));
+        let result = process(table_name, span, &db, columns);
         // SQLite uses dynamic typing, making any length acceptable for a varchar column
         assert!(result.is_ok());
     }
@@ -251,7 +305,7 @@ mod test {
             Value::test_string("ThisIsTheWrongType".to_string()),
         );
 
-        let result = process(table_name, span, &db, Some(columns));
+        let result = process(table_name, span, &db, columns);
         // SQLite uses dynamic typing, making any type acceptable for a column
         assert!(result.is_ok());
     }
@@ -276,7 +330,7 @@ mod test {
             Value::test_string("ThisIsALongString".to_string()),
         );
 
-        let result = process(table_name, span, &db, Some(columns));
+        let result = process(table_name, span, &db, columns);
 
         assert!(result.is_err());
     }
@@ -293,7 +347,7 @@ mod test {
             Value::test_string("ThisIsALongString".to_string()),
         );
 
-        let result = process(table_name, span, &db, Some(columns));
+        let result = process(table_name, span, &db, columns);
 
         assert!(result.is_err());
     }

crates/nu-command/src/stor/update.rs

@@ -11,14 +11,17 @@ impl Command for StorUpdate {
 
     fn signature(&self) -> Signature {
         Signature::build("stor update")
-            .input_output_types(vec![(Type::Nothing, Type::table())])
+            .input_output_types(vec![
+                (Type::Nothing, Type::table()),
+                (Type::record(), Type::table()),
+            ])
             .required_named(
                 "table-name",
                 SyntaxShape::String,
                 "name of the table you want to insert into",
                 Some('t'),
             )
-            .required_named(
+            .named(
                 "update-record",
                 SyntaxShape::Record(vec![]),
                 "a record of column names and column values to update in the specified table",
@@ -54,6 +57,11 @@ impl Command for StorUpdate {
             example: "stor update --table-name nudb --update-record {str1: nushell datetime1: 2020-04-17} --where-clause \"bool1 = 1\"",
             result: None,
         },
+        Example {
+            description: "Update the in-memory sqlite database through pipeline input",
+            example: "{str1: nushell datetime1: 2020-04-17} | stor update --table-name nudb",
+            result: None,
+        },
         ]
     }
 
@@ -62,91 +70,147 @@ impl Command for StorUpdate {
         engine_state: &EngineState,
         stack: &mut Stack,
         call: &Call,
-        _input: PipelineData,
+        input: PipelineData,
     ) -> Result<PipelineData, ShellError> {
         let span = call.head;
         let table_name: Option<String> = call.get_flag(engine_state, stack, "table-name")?;
-        let columns: Option<Record> = call.get_flag(engine_state, stack, "update-record")?;
+        let update_record: Option<Record> = call.get_flag(engine_state, stack, "update-record")?;
         let where_clause_opt: Option<Spanned<String>> =
             call.get_flag(engine_state, stack, "where-clause")?;
 
         // Open the in-mem database
         let db = Box::new(SQLiteDatabase::new(std::path::Path::new(MEMORY_DB), None));
 
-        if table_name.is_none() {
-            return Err(ShellError::MissingParameter {
-                param_name: "requires at table name".into(),
-                span,
-            });
-        }
-        let new_table_name = table_name.unwrap_or("table".into());
-        if let Ok(conn) = db.open_connection() {
-            match columns {
-                Some(record) => {
-                    let mut update_stmt = format!("UPDATE {} ", new_table_name);
+        // Check if the record is being passed as input or using the update record parameter
+        let columns = handle(span, update_record, input)?;
 
-                    update_stmt.push_str("SET ");
-                    let vals = record.iter();
-                    vals.for_each(|(key, val)| match val {
-                        Value::Int { val, .. } => {
-                            update_stmt.push_str(&format!("{} = {}, ", key, val));
-                        }
-                        Value::Float { val, .. } => {
-                            update_stmt.push_str(&format!("{} = {}, ", key, val));
-                        }
-                        Value::String { val, .. } => {
-                            update_stmt.push_str(&format!("{} = '{}', ", key, val));
-                        }
-                        Value::Date { val, .. } => {
-                            update_stmt.push_str(&format!("{} = '{}', ", key, val));
-                        }
-                        Value::Bool { val, .. } => {
-                            update_stmt.push_str(&format!("{} = {}, ", key, val));
-                        }
-                        _ => {
-                            // return Err(ShellError::UnsupportedInput {
-                            //     msg: format!("{} is not a valid datepart, expected one of year, month, day, hour, minute, second, millisecond, microsecond, nanosecond", part.item),
-                            //     input: "value originates from here".to_string(),
-                            //     msg_span: span,
-                            //     input_span: val.span(),
-                            // });
-                        }
-                    });
-                    if update_stmt.ends_with(", ") {
-                        update_stmt.pop();
-                        update_stmt.pop();
-                    }
+        process(table_name, span, &db, columns, where_clause_opt)?;
 
-                    // Yup, this is a bit janky, but I'm not sure a better way to do this without having
-                    // --and and --or flags as well as supporting ==, !=, <>, is null, is not null, etc.
-                    // and other sql syntax. So, for now, just type a sql where clause as a string.
-                    if let Some(where_clause) = where_clause_opt {
-                        update_stmt.push_str(&format!(" WHERE {}", where_clause.item));
-                    }
-                    // dbg!(&update_stmt);
-
-                    conn.execute(&update_stmt, [])
-                        .map_err(|err| ShellError::GenericError {
-                            error: "Failed to open SQLite connection in memory from update".into(),
-                            msg: err.to_string(),
-                            span: Some(Span::test_data()),
-                            help: None,
-                            inner: vec![],
-                        })?;
-                }
-                None => {
-                    return Err(ShellError::MissingParameter {
-                        param_name: "requires at least one column".into(),
-                        span: call.head,
-                    });
-                }
-            };
-        }
-        // dbg!(db.clone());
         Ok(Value::custom(db, span).into_pipeline_data())
     }
 }
 
+fn handle(
+    span: Span,
+    update_record: Option<Record>,
+    input: PipelineData,
+) -> Result<Record, ShellError> {
+    match input {
+        PipelineData::Empty => update_record.ok_or_else(|| ShellError::MissingParameter {
+            param_name: "requires a record".into(),
+            span,
+        }),
+        PipelineData::Value(value, ..) => {
+            // Since input is being used, check if the data record parameter is used too
+            if update_record.is_some() {
+                return Err(ShellError::GenericError {
+                    error: "Pipeline and Flag both being used".into(),
+                    msg: "Use either pipeline input or '--update-record' parameter".into(),
+                    span: Some(span),
+                    help: None,
+                    inner: vec![],
+                });
+            }
+            match value {
+                Value::Record { val, .. } => Ok(val.into_owned()),
+                val => Err(ShellError::OnlySupportsThisInputType {
+                    exp_input_type: "record".into(),
+                    wrong_type: val.get_type().to_string(),
+                    dst_span: Span::unknown(),
+                    src_span: val.span(),
+                }),
+            }
+        }
+        _ => {
+            if update_record.is_some() {
+                return Err(ShellError::GenericError {
+                    error: "Pipeline and Flag both being used".into(),
+                    msg: "Use either pipeline input or '--update-record' parameter".into(),
+                    span: Some(span),
+                    help: None,
+                    inner: vec![],
+                });
+            }
+            Err(ShellError::OnlySupportsThisInputType {
+                exp_input_type: "record".into(),
+                wrong_type: "".into(),
+                dst_span: span,
+                src_span: span,
+            })
+        }
+    }
+}
+
+fn process(
+    table_name: Option<String>,
+    span: Span,
+    db: &SQLiteDatabase,
+    record: Record,
+    where_clause_opt: Option<Spanned<String>>,
+) -> Result<(), ShellError> {
+    if table_name.is_none() {
+        return Err(ShellError::MissingParameter {
+            param_name: "requires at table name".into(),
+            span,
+        });
+    }
+    let new_table_name = table_name.unwrap_or("table".into());
+    if let Ok(conn) = db.open_connection() {
+        let mut update_stmt = format!("UPDATE {} ", new_table_name);
+
+        update_stmt.push_str("SET ");
+        let vals = record.iter();
+        vals.for_each(|(key, val)| match val {
+            Value::Int { val, .. } => {
+                update_stmt.push_str(&format!("{} = {}, ", key, val));
+            }
+            Value::Float { val, .. } => {
+                update_stmt.push_str(&format!("{} = {}, ", key, val));
+            }
+            Value::String { val, .. } => {
+                update_stmt.push_str(&format!("{} = '{}', ", key, val));
+            }
+            Value::Date { val, .. } => {
+                update_stmt.push_str(&format!("{} = '{}', ", key, val));
+            }
+            Value::Bool { val, .. } => {
+                update_stmt.push_str(&format!("{} = {}, ", key, val));
+            }
+            _ => {
+                // return Err(ShellError::UnsupportedInput {
+                //     msg: format!("{} is not a valid datepart, expected one of year, month, day, hour, minute, second, millisecond, microsecond, nanosecond", part.item),
+                //     input: "value originates from here".to_string(),
+                //     msg_span: span,
+                //     input_span: val.span(),
+                // });
+            }
+        });
+        if update_stmt.ends_with(", ") {
+            update_stmt.pop();
+            update_stmt.pop();
+        }
+
+        // Yup, this is a bit janky, but I'm not sure a better way to do this without having
+        // --and and --or flags as well as supporting ==, !=, <>, is null, is not null, etc.
+        // and other sql syntax. So, for now, just type a sql where clause as a string.
+        if let Some(where_clause) = where_clause_opt {
+            update_stmt.push_str(&format!(" WHERE {}", where_clause.item));
+        }
+        // dbg!(&update_stmt);
+
+        conn.execute(&update_stmt, [])
+            .map_err(|err| ShellError::GenericError {
+                error: "Failed to open SQLite connection in memory from update".into(),
+                msg: err.to_string(),
+                span: Some(Span::test_data()),
+                help: None,
+                inner: vec![],
+            })?;
+    }
+    // dbg!(db.clone());
+    Ok(())
+}
+
 #[cfg(test)]
 mod test {
     use super::*;