separation of workflow utils from the main utils

pyaction/utils.py

@@ -1,8 +1,6 @@
 import inspect
 from typing import Callable
 
-from rich.table import Table
-
 from pyaction.exceptions import NotAnnotated
 from pyaction.logger import setup_logger
 
@@ -31,40 +29,3 @@ def check_parameters(func: Callable) -> None:
             raise NotAnnotated(
                 f"Parameter `{param_name}` in the action function `{func.__name__}` is not annotated."
             )
-
-
-def create_output_table() -> Table:
-    """returns a `rich.Table` output table
-
-    Returns:
-        Table: rich table with static columns
-    """
-
-    table = Table(show_lines=True)
-
-    table.add_column(
-        "Variable",
-        justify="center",
-        style="blue bold",
-        vertical="middle",
-    )
-    table.add_column(
-        "Value",
-        justify="center",
-        style="green",
-        vertical="middle",
-    )
-    table.add_column(
-        "Type",
-        justify="center",
-        style="rgb(249,38,114) italic",
-        vertical="middle",
-    )
-    table.add_column(
-        "Workflow Usage",
-        justify="center",
-        style="magenta",
-        vertical="middle",
-    )
-
-    return table

pyaction/workflow/__init__.py

@@ -0,0 +1 @@
+from pyaction.workflow.wrapper import PyAction  # noqa

pyaction/workflow/consts.py

@@ -0,0 +1,19 @@
+# Local stream output table fields
+LOCAL_TABLE_COLS = [
+    {
+        "header": "Variable",
+        "style": "blue bold",
+    },
+    {
+        "header": "Value",
+        "style": "green",
+    },
+    {
+        "header": "Type",
+        "style": "rgb(249,38,114) italic",
+    },
+    {
+        "header": "Workflow Usage",
+        "style": "magenta",
+    },
+]

pyaction/workflow/stream.py

@@ -0,0 +1,58 @@
+from abc import ABC, abstractmethod
+from typing import Any, Dict
+
+from pyaction.consts import GITHUB_OUTPUT, MULTILINE_OUTPUT
+from pyaction.workflow.consts import LOCAL_TABLE_COLS
+
+
+class Stream(ABC):
+    @abstractmethod
+    def put(self, context: Dict[str, Any]):
+        pass
+
+
+class LocalStream(Stream):
+    def __init__(self) -> None:
+        """initializer
+
+        Args:
+            context (Dict[str, Any]): context containing the variables
+        """
+
+        from rich.table import Table
+
+        self.table = Table(show_lines=True)
+        for col in LOCAL_TABLE_COLS:
+            self.table.add_column(**col, justify="center", vertical="middle")
+
+    def put(self, context: Dict[str, Any]):
+        """uses `rich` to put the information into the STDOUT"""
+
+        from rich.console import Console
+
+        console = Console()
+
+        for var, val in context.items():
+            self.table.add_row(
+                var,
+                str(val),
+                str(type(val)),
+                f"${{{{ steps.STEP_ID.outputs.{var} }}}}",
+            )
+
+        console.print(self.table)
+
+
+class WorkflowStream(Stream):
+    def __init__(self, stream: str = GITHUB_OUTPUT) -> None:
+        self.stream = stream
+
+    def put(self, context: Dict[str, Any]):
+        """writes the `context` into the `GITHUB_OUTPUT` environment variable"""
+
+        with open(self.stream, "+w") as streamline:
+            for var, val in context.items():
+                if "\n" in val:
+                    streamline.write(MULTILINE_OUTPUT.format(variable=var, value=val))
+                else:
+                    streamline.write(f"{var}={val}\r\n")