Use pipeline data for http post|put|patch|delete commands. (#13254)

# Description
Provides the ability to use http commands as part of a pipeline.
Additionally, this pull requests extends the pipeline metadata to add a
content_type field. The content_type metadata field allows commands such
as `to json` to set the metadata in the pipeline allowing the http
commands to use it when making requests.

This pull request also introduces the ability to directly stream http
requests from streaming pipelines.

One other small change is that Content-Type will always be set if it is
passed in to the http commands, either indirectly or throw the content
type flag. Previously it was not preserved with requests that were not
of type json or form data.

# User-Facing Changes
* `http post`, `http put`, `http patch`, `http delete` can be used as
part of a pipeline
* `to text`, `to json`, `from json` all set the content_type metadata
field and the http commands will utilize them when making requests.

crates/nu-cmd-extra/src/extra/formats/to/html.rs

@@ -368,6 +368,7 @@ fn theme_demo(span: Span) -> PipelineData {
         .collect();
     Value::list(result, span).into_pipeline_data_with_metadata(PipelineMetadata {
         data_source: DataSource::HtmlThemes,
+        content_type: None,
     })
 }
 

crates/nu-cmd-lang/src/core_commands/collect.rs

@@ -50,6 +50,7 @@ is particularly large, this can cause high memory usage."#
             // check where some input came from.
             Some(PipelineMetadata {
                 data_source: DataSource::FilePath(_),
+                content_type: None,
             }) => None,
             other => other,
         };

crates/nu-command/src/debug/metadata.rs

@@ -80,16 +80,23 @@ impl Command for Metadata {
                     match x {
                         PipelineMetadata {
                             data_source: DataSource::Ls,
+                            ..
                         } => record.push("source", Value::string("ls", head)),
                         PipelineMetadata {
                             data_source: DataSource::HtmlThemes,
+                            ..
                         } => record.push("source", Value::string("into html --list", head)),
                         PipelineMetadata {
                             data_source: DataSource::FilePath(path),
+                            ..
                         } => record.push(
                             "source",
                             Value::string(path.to_string_lossy().to_string(), head),
                         ),
+                        _ => {}
+                    }
+                    if let Some(ref content_type) = x.content_type {
+                        record.push("content_type", Value::string(content_type, head));
                     }
                 }
 
@@ -133,16 +140,23 @@ fn build_metadata_record(arg: &Value, metadata: Option<&PipelineMetadata>, head:
         match x {
             PipelineMetadata {
                 data_source: DataSource::Ls,
+                ..
             } => record.push("source", Value::string("ls", head)),
             PipelineMetadata {
                 data_source: DataSource::HtmlThemes,
+                ..
             } => record.push("source", Value::string("into html --list", head)),
             PipelineMetadata {
                 data_source: DataSource::FilePath(path),
+                ..
             } => record.push(
                 "source",
                 Value::string(path.to_string_lossy().to_string(), head),
             ),
+            _ => {}
+        }
+        if let Some(ref content_type) = x.content_type {
+            record.push("content_type", Value::string(content_type, head));
         }
     }
 

crates/nu-command/src/debug/metadata_set.rs

@@ -46,6 +46,7 @@ impl Command for MetadataSet {
             (Some(path), false) => {
                 let metadata = PipelineMetadata {
                     data_source: DataSource::FilePath(path.into()),
+                    content_type: None,
                 };
                 Ok(input.into_pipeline_data_with_metadata(
                     head,
@@ -56,6 +57,7 @@ impl Command for MetadataSet {
             (None, true) => {
                 let metadata = PipelineMetadata {
                     data_source: DataSource::Ls,
+                    content_type: None,
                 };
                 Ok(input.into_pipeline_data_with_metadata(
                     head,

crates/nu-command/src/filesystem/ls.rs

@@ -122,6 +122,7 @@ impl Command for Ls {
                     ctrl_c,
                     PipelineMetadata {
                         data_source: DataSource::Ls,
+                        content_type: None,
                     },
                 )),
             Some(pattern) => {
@@ -145,6 +146,7 @@ impl Command for Ls {
                         ctrl_c,
                         PipelineMetadata {
                             data_source: DataSource::Ls,
+                            content_type: None,
                         },
                     ))
             }

crates/nu-command/src/filesystem/open.rs

@@ -147,6 +147,7 @@ impl Command for Open {
                         ByteStream::file(file, call_span, ctrlc.clone()),
                         Some(PipelineMetadata {
                             data_source: DataSource::FilePath(path.to_path_buf()),
+                            content_type: None,
                         }),
                     );
 

crates/nu-command/src/formats/from/json.rs

@@ -4,7 +4,7 @@ use std::{
 };
 
 use nu_engine::command_prelude::*;
-use nu_protocol::ListStream;
+use nu_protocol::{ListStream, PipelineMetadata};
 
 #[derive(Clone)]
 pub struct FromJson;
@@ -81,7 +81,7 @@ impl Command for FromJson {
                 PipelineData::Value(Value::String { val, .. }, metadata) => {
                     Ok(PipelineData::ListStream(
                         read_json_lines(Cursor::new(val), span, strict, engine_state.ctrlc.clone()),
-                        metadata,
+                        update_metadata(metadata),
                     ))
                 }
                 PipelineData::ByteStream(stream, metadata)
@@ -90,7 +90,7 @@ impl Command for FromJson {
                     if let Some(reader) = stream.reader() {
                         Ok(PipelineData::ListStream(
                             read_json_lines(reader, span, strict, None),
-                            metadata,
+                            update_metadata(metadata),
                         ))
                     } else {
                         Ok(PipelineData::Empty)
@@ -113,10 +113,10 @@ impl Command for FromJson {
 
             if strict {
                 Ok(convert_string_to_value_strict(&string_input, span)?
-                    .into_pipeline_data_with_metadata(metadata))
+                    .into_pipeline_data_with_metadata(update_metadata(metadata)))
             } else {
                 Ok(convert_string_to_value(&string_input, span)?
-                    .into_pipeline_data_with_metadata(metadata))
+                    .into_pipeline_data_with_metadata(update_metadata(metadata)))
             }
         }
     }
@@ -263,6 +263,14 @@ fn convert_string_to_value_strict(string_input: &str, span: Span) -> Result<Valu
     }
 }
 
+fn update_metadata(metadata: Option<PipelineMetadata>) -> Option<PipelineMetadata> {
+    metadata
+        .map(|md| md.with_content_type(Some("application/json".into())))
+        .or_else(|| {
+            Some(PipelineMetadata::default().with_content_type(Some("application/json".into())))
+        })
+}
+
 #[cfg(test)]
 mod test {
     use super::*;

crates/nu-command/src/formats/to/json.rs

@@ -1,5 +1,5 @@
 use nu_engine::command_prelude::*;
-use nu_protocol::ast::PathMember;
+use nu_protocol::{ast::PathMember, PipelineMetadata};
 
 #[derive(Clone)]
 pub struct ToJson;
@@ -61,7 +61,12 @@ impl Command for ToJson {
 
         match json_result {
             Ok(serde_json_string) => {
-                Ok(Value::string(serde_json_string, span).into_pipeline_data())
+                let res = Value::string(serde_json_string, span);
+                let metadata = PipelineMetadata {
+                    data_source: nu_protocol::DataSource::None,
+                    content_type: Some("application/json".to_string()),
+                };
+                Ok(PipelineData::Value(res, Some(metadata)))
             }
             _ => Ok(Value::error(
                 ShellError::CantConvert {

crates/nu-command/src/formats/to/text.rs

@@ -1,6 +1,8 @@
 use chrono_humanize::HumanTime;
 use nu_engine::command_prelude::*;
-use nu_protocol::{format_duration, format_filesize_from_conf, ByteStream, Config};
+use nu_protocol::{
+    format_duration, format_filesize_from_conf, ByteStream, Config, PipelineMetadata,
+};
 
 const LINE_ENDING: &str = if cfg!(target_os = "windows") {
     "\r\n"
@@ -37,10 +39,14 @@ impl Command for ToText {
         let input = input.try_expand_range()?;
 
         match input {
-            PipelineData::Empty => Ok(Value::string(String::new(), span).into_pipeline_data()),
+            PipelineData::Empty => Ok(Value::string(String::new(), span)
+                .into_pipeline_data_with_metadata(update_metadata(None))),
             PipelineData::Value(value, ..) => {
                 let str = local_into_string(value, LINE_ENDING, engine_state.get_config());
-                Ok(Value::string(str, span).into_pipeline_data())
+                Ok(
+                    Value::string(str, span)
+                        .into_pipeline_data_with_metadata(update_metadata(None)),
+                )
             }
             PipelineData::ListStream(stream, meta) => {
                 let span = stream.span();
@@ -57,10 +63,12 @@ impl Command for ToText {
                         engine_state.ctrlc.clone(),
                         ByteStreamType::String,
                     ),
-                    meta,
+                    update_metadata(meta),
                 ))
             }
-            PipelineData::ByteStream(stream, meta) => Ok(PipelineData::ByteStream(stream, meta)),
+            PipelineData::ByteStream(stream, meta) => {
+                Ok(PipelineData::ByteStream(stream, update_metadata(meta)))
+            }
         }
     }
 
@@ -124,6 +132,14 @@ fn local_into_string(value: Value, separator: &str, config: &Config) -> String {
     }
 }
 
+fn update_metadata(metadata: Option<PipelineMetadata>) -> Option<PipelineMetadata> {
+    metadata
+        .map(|md| md.with_content_type(Some("text/plain".to_string())))
+        .or_else(|| {
+            Some(PipelineMetadata::default().with_content_type(Some("text/plain".to_string())))
+        })
+}
+
 #[cfg(test)]
 mod test {
     use super::*;

crates/nu-command/src/network/http/client.rs

@@ -180,88 +180,113 @@ impl From<ShellError> for ShellErrorOrRequestError {
     }
 }
 
+#[derive(Debug)]
+pub enum HttpBody {
+    Value(Value),
+    ByteStream(ByteStream),
+    None,
+}
+
+// remove once all commands have been migrated
 pub fn send_request(
     request: Request,
-    body: Option<Value>,
+    http_body: HttpBody,
     content_type: Option<String>,
     ctrl_c: Option<Arc<AtomicBool>>,
 ) -> Result<Response, ShellErrorOrRequestError> {
     let request_url = request.url().to_string();
-    if body.is_none() {
-        return send_cancellable_request(&request_url, Box::new(|| request.call()), ctrl_c);
-    }
-    let body = body.expect("Should never be none.");
 
-    let body_type = match content_type {
+    match http_body {
-        Some(it) if it == "application/json" => BodyType::Json,
+        HttpBody::None => {
-        Some(it) if it == "application/x-www-form-urlencoded" => BodyType::Form,
+            send_cancellable_request(&request_url, Box::new(|| request.call()), ctrl_c)
-        _ => BodyType::Unknown,
-    };
-    match body {
-        Value::Binary { val, .. } => send_cancellable_request(
-            &request_url,
-            Box::new(move || request.send_bytes(&val)),
-            ctrl_c,
-        ),
-        Value::String { .. } if body_type == BodyType::Json => {
-            let data = value_to_json_value(&body)?;
-            send_cancellable_request(&request_url, Box::new(|| request.send_json(data)), ctrl_c)
         }
-        Value::String { val, .. } => send_cancellable_request(
+        HttpBody::ByteStream(byte_stream) => {
-            &request_url,
+            let req = if let Some(content_type) = content_type {
-            Box::new(move || request.send_string(&val)),
+                request.set("Content-Type", &content_type)
-            ctrl_c,
+            } else {
-        ),
+                request
-        Value::Record { .. } if body_type == BodyType::Json => {
-            let data = value_to_json_value(&body)?;
-            send_cancellable_request(&request_url, Box::new(|| request.send_json(data)), ctrl_c)
-        }
-        Value::Record { val, .. } if body_type == BodyType::Form => {
-            let mut data: Vec<(String, String)> = Vec::with_capacity(val.len());
-
-            for (col, val) in val.into_owned() {
-                data.push((col, val.coerce_into_string()?))
-            }
-
-            let request_fn = move || {
-                // coerce `data` into a shape that send_form() is happy with
-                let data = data
-                    .iter()
-                    .map(|(a, b)| (a.as_str(), b.as_str()))
-                    .collect::<Vec<(&str, &str)>>();
-                request.send_form(&data)
             };
-            send_cancellable_request(&request_url, Box::new(request_fn), ctrl_c)
+
+            send_cancellable_request_bytes(&request_url, req, byte_stream, ctrl_c)
         }
-        Value::List { vals, .. } if body_type == BodyType::Form => {
+        HttpBody::Value(body) => {
-            if vals.len() % 2 != 0 {
+            let (body_type, req) = match content_type {
-                return Err(ShellErrorOrRequestError::ShellError(ShellError::IOError {
+                Some(it) if it == "application/json" => (BodyType::Json, request),
+                Some(it) if it == "application/x-www-form-urlencoded" => (BodyType::Form, request),
+                Some(it) => {
+                    let r = request.clone().set("Content-Type", &it);
+                    (BodyType::Unknown, r)
+                }
+                _ => (BodyType::Unknown, request),
+            };
+
+            match body {
+                Value::Binary { val, .. } => send_cancellable_request(
+                    &request_url,
+                    Box::new(move || req.send_bytes(&val)),
+                    ctrl_c,
+                ),
+                Value::String { .. } if body_type == BodyType::Json => {
+                    let data = value_to_json_value(&body)?;
+                    send_cancellable_request(&request_url, Box::new(|| req.send_json(data)), ctrl_c)
+                }
+                Value::String { val, .. } => send_cancellable_request(
+                    &request_url,
+                    Box::new(move || req.send_string(&val)),
+                    ctrl_c,
+                ),
+                Value::Record { .. } if body_type == BodyType::Json => {
+                    let data = value_to_json_value(&body)?;
+                    send_cancellable_request(&request_url, Box::new(|| req.send_json(data)), ctrl_c)
+                }
+                Value::Record { val, .. } if body_type == BodyType::Form => {
+                    let mut data: Vec<(String, String)> = Vec::with_capacity(val.len());
+
+                    for (col, val) in val.into_owned() {
+                        data.push((col, val.coerce_into_string()?))
+                    }
+
+                    let request_fn = move || {
+                        // coerce `data` into a shape that send_form() is happy with
+                        let data = data
+                            .iter()
+                            .map(|(a, b)| (a.as_str(), b.as_str()))
+                            .collect::<Vec<(&str, &str)>>();
+                        req.send_form(&data)
+                    };
+                    send_cancellable_request(&request_url, Box::new(request_fn), ctrl_c)
+                }
+                Value::List { vals, .. } if body_type == BodyType::Form => {
+                    if vals.len() % 2 != 0 {
+                        return Err(ShellErrorOrRequestError::ShellError(ShellError::IOError {
+                            msg: "unsupported body input".into(),
+                        }));
+                    }
+
+                    let data = vals
+                        .chunks(2)
+                        .map(|it| Ok((it[0].coerce_string()?, it[1].coerce_string()?)))
+                        .collect::<Result<Vec<(String, String)>, ShellErrorOrRequestError>>()?;
+
+                    let request_fn = move || {
+                        // coerce `data` into a shape that send_form() is happy with
+                        let data = data
+                            .iter()
+                            .map(|(a, b)| (a.as_str(), b.as_str()))
+                            .collect::<Vec<(&str, &str)>>();
+                        req.send_form(&data)
+                    };
+                    send_cancellable_request(&request_url, Box::new(request_fn), ctrl_c)
+                }
+                Value::List { .. } if body_type == BodyType::Json => {
+                    let data = value_to_json_value(&body)?;
+                    send_cancellable_request(&request_url, Box::new(|| req.send_json(data)), ctrl_c)
+                }
+                _ => Err(ShellErrorOrRequestError::ShellError(ShellError::IOError {
                     msg: "unsupported body input".into(),
-                }));
+                })),
             }
-
-            let data = vals
-                .chunks(2)
-                .map(|it| Ok((it[0].coerce_string()?, it[1].coerce_string()?)))
-                .collect::<Result<Vec<(String, String)>, ShellErrorOrRequestError>>()?;
-
-            let request_fn = move || {
-                // coerce `data` into a shape that send_form() is happy with
-                let data = data
-                    .iter()
-                    .map(|(a, b)| (a.as_str(), b.as_str()))
-                    .collect::<Vec<(&str, &str)>>();
-                request.send_form(&data)
-            };
-            send_cancellable_request(&request_url, Box::new(request_fn), ctrl_c)
         }
-        Value::List { .. } if body_type == BodyType::Json => {
-            let data = value_to_json_value(&body)?;
-            send_cancellable_request(&request_url, Box::new(|| request.send_json(data)), ctrl_c)
-        }
-        _ => Err(ShellErrorOrRequestError::ShellError(ShellError::IOError {
-            msg: "unsupported body input".into(),
-        })),
     }
 }
 
@@ -305,6 +330,61 @@ fn send_cancellable_request(
     }
 }
 
+// Helper method used to make blocking HTTP request calls cancellable with ctrl+c
+// ureq functions can block for a long time (default 30s?) while attempting to make an HTTP connection
+fn send_cancellable_request_bytes(
+    request_url: &str,
+    request: Request,
+    byte_stream: ByteStream,
+    ctrl_c: Option<Arc<AtomicBool>>,
+) -> Result<Response, ShellErrorOrRequestError> {
+    let (tx, rx) = mpsc::channel::<Result<Response, ShellErrorOrRequestError>>();
+    let request_url_string = request_url.to_string();
+
+    // Make the blocking request on a background thread...
+    std::thread::Builder::new()
+        .name("HTTP requester".to_string())
+        .spawn(move || {
+            let ret = byte_stream
+                .reader()
+                .ok_or_else(|| {
+                    ShellErrorOrRequestError::ShellError(ShellError::GenericError {
+                        error: "Could not read byte stream".to_string(),
+                        msg: "".into(),
+                        span: None,
+                        help: None,
+                        inner: vec![],
+                    })
+                })
+                .and_then(|reader| {
+                    request.send(reader).map_err(|e| {
+                        ShellErrorOrRequestError::RequestError(request_url_string, Box::new(e))
+                    })
+                });
+
+            // may fail if the user has cancelled the operation
+            let _ = tx.send(ret);
+        })
+        .map_err(ShellError::from)?;
+
+    // ...and poll the channel for responses
+    loop {
+        if nu_utils::ctrl_c::was_pressed(&ctrl_c) {
+            // Return early and give up on the background thread. The connection will either time out or be disconnected
+            return Err(ShellErrorOrRequestError::ShellError(
+                ShellError::InterruptedByUser { span: None },
+            ));
+        }
+
+        // 100ms wait time chosen arbitrarily
+        match rx.recv_timeout(Duration::from_millis(100)) {
+            Ok(result) => return result,
+            Err(RecvTimeoutError::Timeout) => continue,
+            Err(RecvTimeoutError::Disconnected) => panic!("http response channel disconnected"),
+        }
+    }
+}
+
 pub fn request_set_timeout(
     timeout: Option<Value>,
     mut request: Request,

crates/nu-command/src/network/http/delete.rs

@@ -1,7 +1,7 @@
 use crate::network::http::client::{
     check_response_redirection, http_client, http_parse_redirect_mode, http_parse_url,
     request_add_authorization_header, request_add_custom_headers, request_handle_response,
-    request_set_timeout, send_request, RequestFlags,
+    request_set_timeout, send_request, HttpBody, RequestFlags,
 };
 use nu_engine::command_prelude::*;
 
@@ -15,7 +15,7 @@ impl Command for SubCommand {
 
     fn signature(&self) -> Signature {
         Signature::build("http delete")
-            .input_output_types(vec![(Type::Nothing, Type::Any)])
+            .input_output_types(vec![(Type::Any, Type::Any)])
             .allow_variants_without_examples(true)
             .required(
                 "URL",
@@ -132,6 +132,11 @@ impl Command for SubCommand {
                     "http delete --content-type application/json --data { field: value } https://www.example.com",
                 result: None,
             },
+            Example {
+                description: "Perform an HTTP delete with JSON content from a pipeline to example.com",
+                example: "open foo.json | http delete https://www.example.com",
+                result: None,
+            },
         ]
     }
 }
@@ -139,7 +144,7 @@ impl Command for SubCommand {
 struct Arguments {
     url: Value,
     headers: Option<Value>,
-    data: Option<Value>,
+    data: HttpBody,
     content_type: Option<String>,
     raw: bool,
     insecure: bool,
@@ -155,13 +160,27 @@ fn run_delete(
     engine_state: &EngineState,
     stack: &mut Stack,
     call: &Call,
-    _input: PipelineData,
+    input: PipelineData,
 ) -> Result<PipelineData, ShellError> {
+    let (data, maybe_metadata) = call
+        .get_flag::<Value>(engine_state, stack, "data")?
+        .map(|v| (HttpBody::Value(v), None))
+        .unwrap_or_else(|| match input {
+            PipelineData::Value(v, metadata) => (HttpBody::Value(v), metadata),
+            PipelineData::ByteStream(byte_stream, metadata) => {
+                (HttpBody::ByteStream(byte_stream), metadata)
+            }
+            _ => (HttpBody::None, None),
+        });
+    let content_type = call
+        .get_flag(engine_state, stack, "content-type")?
+        .or_else(|| maybe_metadata.and_then(|m| m.content_type));
+
     let args = Arguments {
         url: call.req(engine_state, stack, 0)?,
         headers: call.get_flag(engine_state, stack, "headers")?,
-        data: call.get_flag(engine_state, stack, "data")?,
+        data,
-        content_type: call.get_flag(engine_state, stack, "content-type")?,
+        content_type,
         raw: call.has_flag(engine_state, stack, "raw")?,
         insecure: call.has_flag(engine_state, stack, "insecure")?,
         user: call.get_flag(engine_state, stack, "user")?,

crates/nu-command/src/network/http/get.rs

@@ -5,6 +5,8 @@ use crate::network::http::client::{
 };
 use nu_engine::command_prelude::*;
 
+use super::client::HttpBody;
+
 #[derive(Clone)]
 pub struct SubCommand;
 
@@ -180,7 +182,7 @@ fn helper(
     request = request_add_authorization_header(args.user, args.password, request);
     request = request_add_custom_headers(args.headers, request)?;
 
-    let response = send_request(request.clone(), None, None, ctrl_c);
+    let response = send_request(request.clone(), HttpBody::None, None, ctrl_c);
 
     let request_flags = RequestFlags {
         raw: args.raw,

crates/nu-command/src/network/http/head.rs

@@ -7,6 +7,8 @@ use nu_engine::command_prelude::*;
 
 use std::sync::{atomic::AtomicBool, Arc};
 
+use super::client::HttpBody;
+
 #[derive(Clone)]
 pub struct SubCommand;
 
@@ -156,7 +158,7 @@ fn helper(
     request = request_add_authorization_header(args.user, args.password, request);
     request = request_add_custom_headers(args.headers, request)?;
 
-    let response = send_request(request, None, None, ctrlc);
+    let response = send_request(request, HttpBody::None, None, ctrlc);
     check_response_redirection(redirect_mode, span, &response)?;
     request_handle_response_headers(span, response)
 }

crates/nu-command/src/network/http/options.rs

@@ -4,6 +4,8 @@ use crate::network::http::client::{
 };
 use nu_engine::command_prelude::*;
 
+use super::client::HttpBody;
+
 #[derive(Clone)]
 pub struct SubCommand;
 
@@ -159,7 +161,7 @@ fn helper(
     request = request_add_authorization_header(args.user, args.password, request);
     request = request_add_custom_headers(args.headers, request)?;
 
-    let response = send_request(request.clone(), None, None, ctrl_c);
+    let response = send_request(request.clone(), HttpBody::None, None, ctrl_c);
 
     // http options' response always showed in header, so we set full to true.
     // And `raw` is useless too because options method doesn't return body, here we set to true

crates/nu-command/src/network/http/patch.rs

@@ -1,7 +1,7 @@
 use crate::network::http::client::{
     check_response_redirection, http_client, http_parse_redirect_mode, http_parse_url,
     request_add_authorization_header, request_add_custom_headers, request_handle_response,
-    request_set_timeout, send_request, RequestFlags,
+    request_set_timeout, send_request, HttpBody, RequestFlags,
 };
 use nu_engine::command_prelude::*;
 
@@ -15,10 +15,10 @@ impl Command for SubCommand {
 
     fn signature(&self) -> Signature {
         Signature::build("http patch")
-            .input_output_types(vec![(Type::Nothing, Type::Any)])
+            .input_output_types(vec![(Type::Any, Type::Any)])
             .allow_variants_without_examples(true)
             .required("URL", SyntaxShape::String, "The URL to post to.")
-            .required("data", SyntaxShape::Any, "The contents of the post body.")
+            .optional("data", SyntaxShape::Any, "The contents of the post body.")
             .named(
                 "user",
                 SyntaxShape::Any,
@@ -124,6 +124,11 @@ impl Command for SubCommand {
                 example: "http patch --content-type application/json https://www.example.com { field: value }",
                 result: None,
             },
+            Example {
+                description: "Patch JSON content from a pipeline to example.com",
+                example: "open foo.json | http patch https://www.example.com",
+                result: None,
+            },
         ]
     }
 }
@@ -131,7 +136,7 @@ impl Command for SubCommand {
 struct Arguments {
     url: Value,
     headers: Option<Value>,
-    data: Value,
+    data: HttpBody,
     content_type: Option<String>,
     raw: bool,
     insecure: bool,
@@ -147,13 +152,37 @@ fn run_patch(
     engine_state: &EngineState,
     stack: &mut Stack,
     call: &Call,
-    _input: PipelineData,
+    input: PipelineData,
 ) -> Result<PipelineData, ShellError> {
+    let (data, maybe_metadata) = call
+        .opt::<Value>(engine_state, stack, 1)?
+        .map(|v| (HttpBody::Value(v), None))
+        .unwrap_or_else(|| match input {
+            PipelineData::Value(v, metadata) => (HttpBody::Value(v), metadata),
+            PipelineData::ByteStream(byte_stream, metadata) => {
+                (HttpBody::ByteStream(byte_stream), metadata)
+            }
+            _ => (HttpBody::None, None),
+        });
+    let content_type = call
+        .get_flag(engine_state, stack, "content-type")?
+        .or_else(|| maybe_metadata.and_then(|m| m.content_type));
+
+    if let HttpBody::None = data {
+        return Err(ShellError::GenericError {
+            error: "Data must be provided either through pipeline or positional argument".into(),
+            msg: "".into(),
+            span: Some(call.head),
+            help: None,
+            inner: vec![],
+        });
+    }
+
     let args = Arguments {
         url: call.req(engine_state, stack, 0)?,
         headers: call.get_flag(engine_state, stack, "headers")?,
-        data: call.req(engine_state, stack, 1)?,
+        data,
-        content_type: call.get_flag(engine_state, stack, "content-type")?,
+        content_type,
         raw: call.has_flag(engine_state, stack, "raw")?,
         insecure: call.has_flag(engine_state, stack, "insecure")?,
         user: call.get_flag(engine_state, stack, "user")?,
@@ -187,7 +216,7 @@ fn helper(
     request = request_add_authorization_header(args.user, args.password, request);
     request = request_add_custom_headers(args.headers, request)?;
 
-    let response = send_request(request.clone(), Some(args.data), args.content_type, ctrl_c);
+    let response = send_request(request.clone(), args.data, args.content_type, ctrl_c);
 
     let request_flags = RequestFlags {
         raw: args.raw,

crates/nu-command/src/network/http/post.rs

@@ -1,7 +1,7 @@
 use crate::network::http::client::{
     check_response_redirection, http_client, http_parse_redirect_mode, http_parse_url,
     request_add_authorization_header, request_add_custom_headers, request_handle_response,
-    request_set_timeout, send_request, RequestFlags,
+    request_set_timeout, send_request, HttpBody, RequestFlags,
 };
 use nu_engine::command_prelude::*;
 
@@ -15,10 +15,10 @@ impl Command for SubCommand {
 
     fn signature(&self) -> Signature {
         Signature::build("http post")
-            .input_output_types(vec![(Type::Nothing, Type::Any)])
+            .input_output_types(vec![(Type::Any, Type::Any)])
             .allow_variants_without_examples(true)
             .required("URL", SyntaxShape::String, "The URL to post to.")
-            .required("data", SyntaxShape::Any, "The contents of the post body.")
+            .optional("data", SyntaxShape::Any, "The contents of the post body. Required unless part of a pipeline.")
             .named(
                 "user",
                 SyntaxShape::Any,
@@ -122,6 +122,11 @@ impl Command for SubCommand {
                 example: "http post --content-type application/json https://www.example.com { field: value }",
                 result: None,
             },
+            Example {
+                description: "Post JSON content from a pipeline to example.com",
+                example: "open foo.json | http post https://www.example.com",
+                result: None,
+            },
         ]
     }
 }
@@ -129,7 +134,7 @@ impl Command for SubCommand {
 struct Arguments {
     url: Value,
     headers: Option<Value>,
-    data: Value,
+    data: HttpBody,
     content_type: Option<String>,
     raw: bool,
     insecure: bool,
@@ -145,13 +150,37 @@ fn run_post(
     engine_state: &EngineState,
     stack: &mut Stack,
     call: &Call,
-    _input: PipelineData,
+    input: PipelineData,
 ) -> Result<PipelineData, ShellError> {
+    let (data, maybe_metadata) = call
+        .opt::<Value>(engine_state, stack, 1)?
+        .map(|v| (HttpBody::Value(v), None))
+        .unwrap_or_else(|| match input {
+            PipelineData::Value(v, metadata) => (HttpBody::Value(v), metadata),
+            PipelineData::ByteStream(byte_stream, metadata) => {
+                (HttpBody::ByteStream(byte_stream), metadata)
+            }
+            _ => (HttpBody::None, None),
+        });
+    let content_type = call
+        .get_flag(engine_state, stack, "content-type")?
+        .or_else(|| maybe_metadata.and_then(|m| m.content_type));
+
+    if let HttpBody::None = data {
+        return Err(ShellError::GenericError {
+            error: "Data must be provided either through pipeline or positional argument".into(),
+            msg: "".into(),
+            span: Some(call.head),
+            help: None,
+            inner: vec![],
+        });
+    }
+
     let args = Arguments {
         url: call.req(engine_state, stack, 0)?,
         headers: call.get_flag(engine_state, stack, "headers")?,
-        data: call.req(engine_state, stack, 1)?,
+        data,
-        content_type: call.get_flag(engine_state, stack, "content-type")?,
+        content_type,
         raw: call.has_flag(engine_state, stack, "raw")?,
         insecure: call.has_flag(engine_state, stack, "insecure")?,
         user: call.get_flag(engine_state, stack, "user")?,
@@ -185,7 +214,7 @@ fn helper(
     request = request_add_authorization_header(args.user, args.password, request);
     request = request_add_custom_headers(args.headers, request)?;
 
-    let response = send_request(request.clone(), Some(args.data), args.content_type, ctrl_c);
+    let response = send_request(request.clone(), args.data, args.content_type, ctrl_c);
 
     let request_flags = RequestFlags {
         raw: args.raw,

crates/nu-command/src/network/http/put.rs

@@ -1,7 +1,7 @@
 use crate::network::http::client::{
     check_response_redirection, http_client, http_parse_redirect_mode, http_parse_url,
     request_add_authorization_header, request_add_custom_headers, request_handle_response,
-    request_set_timeout, send_request, RequestFlags,
+    request_set_timeout, send_request, HttpBody, RequestFlags,
 };
 use nu_engine::command_prelude::*;
 
@@ -15,10 +15,10 @@ impl Command for SubCommand {
 
     fn signature(&self) -> Signature {
         Signature::build("http put")
-            .input_output_types(vec![(Type::Nothing, Type::Any)])
+            .input_output_types(vec![(Type::Any, Type::Any)])
             .allow_variants_without_examples(true)
             .required("URL", SyntaxShape::String, "The URL to post to.")
-            .required("data", SyntaxShape::Any, "The contents of the post body.")
+            .optional("data", SyntaxShape::Any, "The contents of the post body. Required unless part of a pipeline.")
             .named(
                 "user",
                 SyntaxShape::Any,
@@ -122,6 +122,11 @@ impl Command for SubCommand {
                 example: "http put --content-type application/json https://www.example.com { field: value }",
                 result: None,
             },
+            Example {
+                description: "Put JSON content from a pipeline to example.com",
+                example: "open foo.json | http put https://www.example.com",
+                result: None,
+            },
         ]
     }
 }
@@ -129,7 +134,7 @@ impl Command for SubCommand {
 struct Arguments {
     url: Value,
     headers: Option<Value>,
-    data: Value,
+    data: HttpBody,
     content_type: Option<String>,
     raw: bool,
     insecure: bool,
@@ -145,13 +150,38 @@ fn run_put(
     engine_state: &EngineState,
     stack: &mut Stack,
     call: &Call,
-    _input: PipelineData,
+    input: PipelineData,
 ) -> Result<PipelineData, ShellError> {
+    let (data, maybe_metadata) = call
+        .opt::<Value>(engine_state, stack, 1)?
+        .map(|v| (HttpBody::Value(v), None))
+        .unwrap_or_else(|| match input {
+            PipelineData::Value(v, metadata) => (HttpBody::Value(v), metadata),
+            PipelineData::ByteStream(byte_stream, metadata) => {
+                (HttpBody::ByteStream(byte_stream), metadata)
+            }
+            _ => (HttpBody::None, None),
+        });
+
+    if let HttpBody::None = data {
+        return Err(ShellError::GenericError {
+            error: "Data must be provided either through pipeline or positional argument".into(),
+            msg: "".into(),
+            span: Some(call.head),
+            help: None,
+            inner: vec![],
+        });
+    }
+
+    let content_type = call
+        .get_flag(engine_state, stack, "content-type")?
+        .or_else(|| maybe_metadata.and_then(|m| m.content_type));
+
     let args = Arguments {
         url: call.req(engine_state, stack, 0)?,
         headers: call.get_flag(engine_state, stack, "headers")?,
-        data: call.req(engine_state, stack, 1)?,
+        data,
-        content_type: call.get_flag(engine_state, stack, "content-type")?,
+        content_type,
         raw: call.has_flag(engine_state, stack, "raw")?,
         insecure: call.has_flag(engine_state, stack, "insecure")?,
         user: call.get_flag(engine_state, stack, "user")?,
@@ -185,7 +215,7 @@ fn helper(
     request = request_add_authorization_header(args.user, args.password, request);
     request = request_add_custom_headers(args.headers, request)?;
 
-    let response = send_request(request.clone(), Some(args.data), args.content_type, ctrl_c);
+    let response = send_request(request.clone(), args.data, args.content_type, ctrl_c);
 
     let request_flags = RequestFlags {
         raw: args.raw,

crates/nu-command/src/viewers/table.rs

@@ -605,6 +605,7 @@ fn handle_row_stream(
         // First, `ls` sources:
         Some(PipelineMetadata {
             data_source: DataSource::Ls,
+            ..
         }) => {
             let config = get_config(input.engine_state, input.stack);
             let ls_colors_env_str = match input.stack.get_env_var(input.engine_state, "LS_COLORS") {
@@ -636,6 +637,7 @@ fn handle_row_stream(
         // Next, `to html -l` sources:
         Some(PipelineMetadata {
             data_source: DataSource::HtmlThemes,
+            ..
         }) => {
             stream.map(|mut value| {
                 if let Value::Record { val: record, .. } = &mut value {

crates/nu-command/tests/commands/network/http/delete.rs

@@ -20,6 +20,25 @@ fn http_delete_is_success() {
     assert!(actual.out.is_empty())
 }
 
+#[test]
+fn http_delete_is_success_pipeline() {
+    let mut server = Server::new();
+
+    let _mock = server.mock("DELETE", "/").create();
+
+    let actual = nu!(pipeline(
+        format!(
+            r#"
+        "foo" | http delete {url}
+        "#,
+            url = server.url()
+        )
+        .as_str()
+    ));
+
+    assert!(actual.out.is_empty())
+}
+
 #[test]
 fn http_delete_failed_due_to_server_error() {
     let mut server = Server::new();

crates/nu-command/tests/commands/network/http/patch.rs

@@ -20,6 +20,25 @@ fn http_patch_is_success() {
     assert!(actual.out.is_empty())
 }
 
+#[test]
+fn http_patch_is_success_pipeline() {
+    let mut server = Server::new();
+
+    let _mock = server.mock("PATCH", "/").match_body("foo").create();
+
+    let actual = nu!(pipeline(
+        format!(
+            r#"
+        "foo" | http patch {url}
+        "#,
+            url = server.url()
+        )
+        .as_str()
+    ));
+
+    assert!(actual.out.is_empty())
+}
+
 #[test]
 fn http_patch_failed_due_to_server_error() {
     let mut server = Server::new();
@@ -55,7 +74,9 @@ fn http_patch_failed_due_to_missing_body() {
         .as_str()
     ));
 
-    assert!(actual.err.contains("Usage: http patch"))
+    assert!(actual
+        .err
+        .contains("Data must be provided either through pipeline or positional argument"))
 }
 
 #[test]

crates/nu-command/tests/commands/network/http/post.rs

@@ -19,6 +19,24 @@ fn http_post_is_success() {
 
     assert!(actual.out.is_empty())
 }
+#[test]
+fn http_post_is_success_pipeline() {
+    let mut server = Server::new();
+
+    let _mock = server.mock("POST", "/").match_body("foo").create();
+
+    let actual = nu!(pipeline(
+        format!(
+            r#"
+        "foo" | http post {url}
+        "#,
+            url = server.url()
+        )
+        .as_str()
+    ));
+
+    assert!(actual.out.is_empty())
+}
 
 #[test]
 fn http_post_failed_due_to_server_error() {
@@ -55,7 +73,9 @@ fn http_post_failed_due_to_missing_body() {
         .as_str()
     ));
 
-    assert!(actual.err.contains("Usage: http post"))
+    assert!(actual
+        .err
+        .contains("Data must be provided either through pipeline or positional argument"))
 }
 
 #[test]

crates/nu-command/tests/commands/network/http/put.rs

@@ -20,6 +20,25 @@ fn http_put_is_success() {
     assert!(actual.out.is_empty())
 }
 
+#[test]
+fn http_put_is_success_pipeline() {
+    let mut server = Server::new();
+
+    let _mock = server.mock("PUT", "/").match_body("foo").create();
+
+    let actual = nu!(pipeline(
+        format!(
+            r#"
+        "foo" | http put {url} 
+        "#,
+            url = server.url()
+        )
+        .as_str()
+    ));
+
+    assert!(actual.out.is_empty())
+}
+
 #[test]
 fn http_put_failed_due_to_server_error() {
     let mut server = Server::new();
@@ -55,7 +74,9 @@ fn http_put_failed_due_to_missing_body() {
         .as_str()
     ));
 
-    assert!(actual.err.contains("Usage: http put"))
+    assert!(actual
+        .err
+        .contains("Data must be provided either through pipeline or positional argument"))
 }
 
 #[test]

crates/nu-plugin-core/src/interface/tests.rs

@@ -18,6 +18,7 @@ use std::{path::Path, sync::Arc};
 fn test_metadata() -> PipelineMetadata {
     PipelineMetadata {
         data_source: DataSource::FilePath("/test/path".into()),
+        content_type: None,
     }
 }
 
@@ -258,7 +259,7 @@ fn read_pipeline_data_prepared_properly() -> Result<(), ShellError> {
     });
     match manager.read_pipeline_data(header, None)? {
         PipelineData::ListStream(_, meta) => match meta {
-            Some(PipelineMetadata { data_source }) => match data_source {
+            Some(PipelineMetadata { data_source, .. }) => match data_source {
                 DataSource::FilePath(path) => {
                     assert_eq!(Path::new("/test/path"), path);
                     Ok(())

crates/nu-protocol/src/pipeline/metadata.rs

@@ -1,18 +1,37 @@
 use std::path::PathBuf;
 
 /// Metadata that is valid for the whole [`PipelineData`](crate::PipelineData)
-#[derive(Debug, Clone)]
+#[derive(Debug, Default, Clone)]
 pub struct PipelineMetadata {
     pub data_source: DataSource,
+    pub content_type: Option<String>,
+}
+
+impl PipelineMetadata {
+    pub fn with_data_source(self, data_source: DataSource) -> Self {
+        Self {
+            data_source,
+            ..self
+        }
+    }
+
+    pub fn with_content_type(self, content_type: Option<String>) -> Self {
+        Self {
+            content_type,
+            ..self
+        }
+    }
 }
 
 /// Describes where the particular [`PipelineMetadata`] originates.
 ///
 /// This can either be a particular family of commands (useful so downstream commands can adjust
 /// the presentation e.g. `Ls`) or the opened file to protect against overwrite-attempts properly.
-#[derive(Debug, Clone)]
+#[derive(Debug, Default, Clone)]
 pub enum DataSource {
     Ls,
     HtmlThemes,
     FilePath(PathBuf),
+    #[default]
+    None,
 }