add shell plugin (#1)

* add shell plugin

Author: grantleehoffman

Makefile

@@ -0,0 +1,40 @@
+SHELL := /bin/bash
+
+.PHONY: setup
+setup: venv lint test
+
+.PHONY: clean-pyc
+clean-pyc:
+	find . -name '*.pyc' -exec rm -rf {} +
+	find . -name '*.pyo' -exec rm -rf {} +
+	find . -name '*~' -exec rm -rf {} +
+	find . -name '__pycache__' -exec rm -rf {} +
+
+.PHONY: clean-test
+clean-test:
+	rm -rf .pytest_cache
+	rm -rf .mypy_cache
+	rm -rf .coverage
+	rm -rf .reports
+	rm -rf htmlcov/
+	rm -rf .pytest_cache
+
+.PHONY: clean
+clean: clean-pyc clean-test
+
+venv: poetry.lock
+	poetry install
+
+.PHONY: format
+format: venv
+	poetry run black nodestream_plugin_shell tests
+	poetry run isort nodestream_plugin_shell tests
+
+.PHONY: lint
+lint: venv
+	poetry run black nodestream_plugin_shell tests --check
+	poetry run ruff nodestream_plugin_shell tests
+
+.PHONY: test
+test: venv
+	poetry run pytest

README.md

@@ -1,2 +1,61 @@
-# nodestream-plugin-shell
-Nodestream Plugin Shell
+# Nodestream Shell Plugin
+
+This plugin provides a [Nodestream](https://github.com/nodestream-proj/nodestream) shell extractor. 
+
+**NOTE: This plugin is currently in development and is not yet ready for production use.**
+
+## Installation
+
+```bash
+pip install nodestream-plugin-shell
+```
+
+## For commands that return output in json format
+```yaml
+# pipeline.yaml
+- implementation: nodestream_plugin_shell:Shell
+  arguments:
+    command: curl
+    arguments:
+      - https://example.com/api/data/json
+    options: 
+      X: GET
+      header: "Accept: application/json"
+    ignore_stdout: false #Provide json output from command to next step in pipeline.
+```
+
+This would be equivalent to running: 
+```
+curl https://example.com/api/data.json -X GET --header "Accept: application/json"
+```
+
+## For commands that create files that you would like to read into the pipeline
+
+#### Terminal command
+
+
+```yaml
+# pipeline.yaml
+- implementation: nodestream_plugin_shell:Shell
+  arguments:
+    command: curl
+    arguments:
+      - https://example.com/api/data.json
+    options: 
+      X: GET
+      header: "Accept: application/json"
+    flags: 
+      - O
+    ignore_stdout: true #Do not provide output from command to next step in pipeline.
+
+- implementation: nodestream.pipeline.extractors:FileExtractor
+  arguments:
+    globs:
+    - data.json
+```
+
+This would be equivalent to running: 
+```
+curl https://example.com/api/data.json -X GET --header "Accept: application/json" -O
+cat data.json
+```

nodestream_plugin_shell/__init__.py

@@ -0,0 +1,3 @@
+from .shell import Shell
+
+__all__ = ("Shell",)

nodestream_plugin_shell/shell.py

@@ -0,0 +1,79 @@
+import json
+import subprocess
+from io import StringIO
+from logging import getLogger
+from typing import Any, AsyncGenerator, Dict, Optional, Set
+
+from nodestream.pipeline import Extractor
+
+
+class Shell(Extractor):
+    """Run shell command and provide output to pipeline"""
+
+    def __init__(
+        self,
+        command: str,
+        arguments: Optional[Set[str]] = None,
+        options: Optional[Dict[str, str]] = None,
+        flags: Optional[Set[str]] = None,
+        ignore_stdout: Optional[bool] = False,
+    ):
+        """Initializes the instance of the extractor.
+        Expects stdout in json format.
+
+        Args:
+            command (str): The command for the execution call.
+            arguments ([str]): The arguments used in the command.
+            options (dict): The Options used in the command.
+            flags ([str]): Flags to use in the command.
+            ignore_stdout (bool): Don't pass stdout to pipeline.
+        """
+        self.command = command
+        self.arguments = arguments if arguments is not None else set()
+        self.options = options if options is not None else dict()
+        self.flags = flags if flags is not None else set()
+        self.ignore_stdout = ignore_stdout
+        self.logger = getLogger(self.__class__.__name__)
+
+    def read_from_file(self) -> StringIO:
+        with open(self.output_file, "r") as file:
+            return StringIO(file.read().strip())
+
+    def build_command(self):
+        cmd = [self.command]
+        cmd.extend(self.arguments)
+        for k, v in self.options.items():
+            if len(k) == 1:
+                cmd.append(f"-{k} {v}")
+            else:
+                cmd.append(f"--{k}={v}")
+        for flag in self.flags:
+            flag = f"-{flag}" if len(flag) == 1 else f"--{flag}"
+            cmd.append(flag)
+        return cmd
+
+    def run_command(self, cmd):
+        process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        stdout, stderr = process.communicate()
+        if process.returncode != 0:
+            raise SystemError(
+                f"Error running shell command: {' '.join(cmd)}, error: {stderr.decode()}, return code: {process.returncode}"
+            )
+        return stdout
+
+    async def extract_records(self) -> AsyncGenerator[Any, Any]:
+        cmd = self.build_command()
+        stdout = self.run_command(cmd)
+        output = stdout.decode().strip()
+        if self.ignore_stdout:
+            yield {}
+        else:
+            try:
+                results = json.loads(output)
+                for item in results:
+                    yield item
+            except json.JSONDecodeError:
+                self.logger.warn(
+                    f"Stdout not in json format, step yielding no data, set ignore_stdout to True to ignore this warning. Stdout: {output}"
+                )
+                yield {}