refactored

pyaction/utils.py

@@ -1,33 +1,5 @@
-import inspect
 import os
-from typing import Callable, Optional
-
-from pyaction.exceptions import NotAnnotated
-from pyaction.logging import Logger
-
-logger = Logger(__name__, include_name=False)
-
-
-def check_parameters(func: Callable) -> None:
-    """
-    Checks the function params typing and annotation.
-
-    Args:
-        func (Callable): The function.
-
-    Raises:
-        NotAnnotated: If there is param(s) not annotated.
-    """
-    signature = inspect.signature(func)
-    for param_name, param in signature.parameters.items():
-        if param.annotation == inspect.Parameter.empty:
-            raise NotAnnotated(f"parameter `{param_name}` is not annotated.")
-
-        if param.default != inspect.Parameter.empty:
-            logger.warning(
-                f"parameter `{param_name}` has a default value which has no effect. "
-                "Set the default value(s) inside the `action.yml`."
-            )
+from typing import Optional
 
 
 def get_running_platform() -> Optional[str]:

pyaction/workflow/utils.py

@@ -0,0 +1,57 @@
+import inspect
+import os
+from typing import Callable
+
+from pyaction.exceptions import NotAnnotated, WorkflowParameterNotFound
+from pyaction.logging import Logger
+
+logger = Logger(__name__, include_name=False)
+
+
+def check_parameters(func: Callable) -> None:
+    """
+    Checks the function params typing and annotation.
+
+    Args:
+        func (Callable): The function.
+
+    Raises:
+        NotAnnotated: If there is param(s) not annotated.
+    """
+    signature = inspect.signature(func)
+    for param_name, param in signature.parameters.items():
+        if param.annotation == inspect.Parameter.empty:
+            raise NotAnnotated(f"parameter `{param_name}` is not annotated.")
+
+        if param.default != inspect.Parameter.empty:
+            logger.warning(
+                f"parameter `{param_name}` has a default value which has no effect. "
+                "Set the default value(s) inside the `action.yml`."
+            )
+
+
+def read_input_parameter(param: str) -> str:
+    """
+    Reads a parameter from the inputs.
+
+    Args:
+        param (str): Parameter name
+
+    Raises:
+        WorkflowParameterNotFound: If the `param` is missing.
+
+    Returns:
+        str: Value of `param`.
+    """
+    prefix = "INPUT_"
+    var_name = prefix + param.upper()
+
+    try:
+        value = os.environ[var_name]
+    except KeyError:
+        raise WorkflowParameterNotFound(
+            f"unable to read the `{var_name}` parameter. "
+            "Make sure it's declared properly."
+        )
+
+    return value

pyaction/workflow/wrapper.py

@@ -3,62 +3,59 @@
 
 from pydantic import TypeAdapter
 
-from pyaction import io
 from pyaction.exceptions import NotAnnotated
 from pyaction.logging import Logger
-from pyaction.utils import check_parameters
-from pyaction.workflow.stream import WorkflowContext
+from pyaction.utils import get_running_platform
+from pyaction.workflow.stream import WorkflowContext, push_to_local, push_to_runner
+from pyaction.workflow.utils import check_parameters, read_input_parameter
 
 logger = Logger(__name__, include_name=False)
 
 
-class Action:
-    def __call__(self, func: Callable) -> None:
-        """
-        Validate parameters based on function annotations and process them accordingly.
-
-        Args:
-            func (Callable): The function to be executed with validated parameters.
-
-        Raises:
-            NotAnnotated: If function parameters are not annotated properly.
-
-        Returns:
-            The result of the function call with validated parameters.
-        """
-        try:
-            check_parameters(func)
-        except NotAnnotated as e:
-            logger.error(str(e))
-            sys.exit(1)
-
-        params = {
-            key: (type_, io.read(key))
-            for key, type_ in get_type_hints(func).items()
-            if key != "return"
-        }
-
-        retyped_params = {}
-
-        for key, item in params.items():
-            retyped_params[key] = TypeAdapter(item[0]).validate_python(item[1])
-
-        return func(**retyped_params)
-
-
 class PyAction:
-    def __init__(self) -> None:
-        """
-        Initializes PyAction class with an instance of the Action class.
-        """
-        self.action = Action
+    def action(self, func: Callable):
+        def wrapper(*args, **kwargs):
+            try:
+                # Validate parameter annotations
+                check_parameters(func)
+            except NotAnnotated as e:
+                logger.error(f"Parameter validation error: {e}")
+                sys.exit(1)
+
+            # Extract parameter types and values
+            param_hints = get_type_hints(func)
+            params = {
+                key: (type_, read_input_parameter(key))
+                for key, type_ in param_hints.items()
+                if key != "return"
+            }
+
+            # Validate and retype parameters
+            try:
+                retyped_params = {
+                    key: TypeAdapter(type_).validate_python(value)
+                    for key, (type_, value) in params.items()
+                }
+            except Exception as e:
+                logger.error(f"Parameter type validation error: {e}")
+                sys.exit(1)
+
+            # Execute the function with retyped parameters
+            func(**retyped_params)
+
+        # Execute the wrapper immediately
+        wrapper()
+        return func
 
     @staticmethod
     def write(context: WorkflowContext) -> None:
         """
-        Writes the `context` env var(s) into the streamline.
+        Writes the context to a stream based on the action running platform.
 
         Args:
-            context (WorkflowContext): Variables and values.
+            context (WorkflowContext): The context to be written.
         """
-        io.write(context)  # pragma: no cover
+        if platform := get_running_platform():
+            push_to_runner(context, stream=platform)
+        else:
+            push_to_local(context)