From add2d11a0206f1d591cf18616f64aeff892fe4f0 Mon Sep 17 00:00:00 2001 From: Sadra Yahyapour Date: Tue, 19 Nov 2024 07:28:04 +0330 Subject: [PATCH] refactored --- pyaction/io.py | 45 ------------------- pyaction/utils.py | 30 +------------ pyaction/workflow/utils.py | 57 +++++++++++++++++++++++ pyaction/workflow/wrapper.py | 87 +++++++++++++++++------------------- 4 files changed, 100 insertions(+), 119 deletions(-) delete mode 100644 pyaction/io.py create mode 100644 pyaction/workflow/utils.py diff --git a/pyaction/io.py b/pyaction/io.py deleted file mode 100644 index 5ac6a1d..0000000 --- a/pyaction/io.py +++ /dev/null @@ -1,45 +0,0 @@ -import os - -from pyaction.exceptions import WorkflowParameterNotFound -from pyaction.utils import get_running_platform -from pyaction.workflow.stream import WorkflowContext, push_to_local, push_to_runner - - -def write(context: WorkflowContext) -> None: - """ - Writes the context to a stream based on the action running platform. - - Args: - context (WorkflowContext): The context to be written. - """ - if platform := get_running_platform(): - push_to_runner(context, stream=platform) - else: - push_to_local(context) - - -def read(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 diff --git a/pyaction/utils.py b/pyaction/utils.py index bfc6e03..f3270bf 100644 --- a/pyaction/utils.py +++ b/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]: diff --git a/pyaction/workflow/utils.py b/pyaction/workflow/utils.py new file mode 100644 index 0000000..cd0206a --- /dev/null +++ b/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 diff --git a/pyaction/workflow/wrapper.py b/pyaction/workflow/wrapper.py index 30f75de..eb9a26e 100644 --- a/pyaction/workflow/wrapper.py +++ b/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)