add HttpRequest command #78 (#79)

mara_pipelines/commands/http.py

@@ -0,0 +1,37 @@
+"""Commands for interacting with HTTP"""
+
+from mara_page import html, _
+from .. import pipelines
+from ..shell import http_request_command
+
+
+class HttpRequest(pipelines.Command):
+    def __init__(self, url: str, headers: {str: str} = None, method: str = None, body: str = None) -> None:
+        """
+        Executes a HTTP request
+
+        Args:
+            url: The url
+            headers: The HTTP headers as dict
+            method: The HTTP method to be used
+            body: The body string to be used in the HTTP request
+        """
+        super().__init__()
+        self.url = url
+        self.headers = headers
+        self.method = method
+        self.body = body
+
+    def shell_command(self):
+        return http_request_command(self.url, self.headers, self.method, self.body)
+
+    def html_doc_items(self) -> [(str, str)]:
+        return [
+            ('method', _.tt[self.method or 'GET']),
+            ('url', _.tt[self.url]),
+            ('headers', _.pre[
+                '\n'.join([f'{header}: {content}' for header, content in self.headers.items()]) if self.headers else ''
+            ]),
+            ('body', _.pre[self.body or '']),
+            ('command', html.highlight_syntax(self.shell_command(), 'bash'))
+        ]

mara_pipelines/shell.py

@@ -1,6 +1,7 @@
 """Command execution in bash shells"""
 
 import time
+import shlex
 
 from . import config
 from .logging import logger
@@ -84,5 +85,30 @@ def quote(s):
            + '"'
 
 
+def http_request_command(url: str, headers: {str: str} = None, method: str = 'GET', body: str = None, body_from_stdin: bool = False) -> str:
+    """
+    Creates a curl command sending a HTTP request
+
+    Args:
+        url: The url
+        headers: The HTTP headers as dict
+        method: The HTTP method to be used
+        body: The body string to be used in the HTTP request
+        body_from_stdin: Read the body for the HTTP request from stdin
+    """
+    if body and body_from_stdin:
+        raise ValueError('You can only use body or body_from_stdin but not both')
+
+    def quote(s):
+        return str(s).replace('\\', '\\\\').replace('"', '\\"')
+
+    return ("curl -sf"
+            + (f' -X {method}' if method and method != 'GET' else '')
+            + (''.join([f' -H "{quote(header)}: {quote(content)}"' for header, content in headers.items()]) if headers else '')
+            + (f' --data {shlex.quote(body)}' if body else '')
+            + (' --data-binary @-' if body_from_stdin else '')
+            + f" {shlex.quote(url)}")
+
+
 if __name__ == "__main__":
     run_shell_command('ping -c 3 google.com; ping null')