Repository: keephq/keep
Pull request: chore: Enhancement: Merge step and action class implementation
<!-- 
Thanks for creating this pull request 🤗

Please make sure that the pull request is limited to one type (docs, feature, etc.) and keep it as small as possible. You can open multiple prs instead of opening a huge one.
-->

<!-- If this pull request closes an issue, please mention the issue number below -->
Closes #180  <!-- Issue # 180 -->

## 📑 Description
<!-- Add a brief description of the pr -->
This PR consolidates the Actions and Steps class. See reason below;
Actions and Steps are practically the same entities, but the fact that Actions has conditions (and we want conditions for steps as well) and "throttling" logic, and Step calls `query` when executed while Actions calls `notify`.

The reason this is now needed is that we want to add an `on-failure` strategy for both Actions and Steps and would like to have the same logic implemented for both. (And in general, this can reduce code and complexity).

<!-- You can also choose to add a list of changes and if they have been completed or not by using the markdown to-do list syntax
- [ ] Not Completed
- [x] Completed
-->

## ✅ Checks
<!-- Make sure your pr passes the CI checks and do check the following fields as needed - -->
- [x] My pull request adheres to the code style of this project
- [ ] My code requires changes to the documentation
- [ ] I have updated the documentation as required
- [x] All the tests have passed

## ℹ Additional Information
<!-- Any additional information like breaking changes, dependencies added, screenshots, comparisons between new and old behavior, etc. -->


Problem statement:
Enhancement: Merge step and action class implementation (e.g. base class)
Actions and Steps are practically the same entities, but the fact that Actions has conditions (and we want conditions for steps as well) and "throttling" logic, and Step calls `query` when executed while Actions calls `notify`.

The reason this is now needed is that we want to add an `on-failure` strategy for both Actions and Steps and would like to have the same logic implemented for both. (And in general, this can reduce code and complexity).

Hints from issue discussion:
@shahargl FYI, if you have thoughts, please add them.
Adding to this issue: in general, we need to re-think the "errors" mechanism in our alerts workflow.
The run alert function is quite messy and is hard to understand what happens when an error occurs or what an error even is:

```python
 def _run_alerts(self, alerts: typing.List[Alert]):
        alerts_errors = []
        for alert in alerts:
            # otherwise any(errors) might throw an exception
            errors = []
            self.logger.info(f"Running alert {alert.alert_id}")
            try:
                errors = alert.run()
            except Exception as e:
                self.logger.error(
                    f"Error running alert {alert.alert_id}", extra={"exception": e}
                )
                raise
            if any(errors):
                self.logger.info(msg=f"Alert {alert.alert_id} ran with errors")
            else:
                self.logger.info(f"Alert {alert.alert_id} ran successfully")
            alerts_errors.extend(errors)
        return alerts_errors
```

Errors can either be raised in some conditions or returned as "errors" from `alert.run()`. Moreover, it is quite confusing that running an alert returns its errors, IMHO.
> Adding to this issue: in general, we need to re-think the "errors" mechanism in our alerts workflow. The run alert function is quite messy and is hard to understand what happens when an error occurs or what an error even is:
> 
> ```python
>  def _run_alerts(self, alerts: typing.List[Alert]):
>         alerts_errors = []
>         for alert in alerts:
>             # otherwise any(errors) might throw an exception
>             errors = []
>             self.logger.info(f"Running alert {alert.alert_id}")
>             try:
>                 errors = alert.run()
>             except Exception as e:
>                 self.logger.error(
>                     f"Error running alert {alert.alert_id}", extra={"exception": e}
>                 )
>                 raise
>             if any(errors):
>                 self.logger.info(msg=f"Alert {alert.alert_id} ran with errors")
>             else:
>                 self.logger.info(f"Alert {alert.alert_id} ran successfully")
>             alerts_errors.extend(errors)
>         return alerts_errors
> ```
> 
> Errors can either be raised in some conditions or returned as "errors" from `alert.run()`. Moreover, it is quite confusing that running an alert returns its errors, IMHO.

Created this temporary on-failure mechanism until we resolve this issue: https://github.com/keephq/keep/pull/182

Commit message: chore(Enhancement) Merge step and action class implementation

Patch to review:
diff --git a/docs/platform/core/syntax/basic-syntax.mdx b/docs/platform/core/syntax/basic-syntax.mdx
index 261d920d94..f3d989bca3 100644
--- a/docs/platform/core/syntax/basic-syntax.mdx
+++ b/docs/platform/core/syntax/basic-syntax.mdx
@@ -53,17 +53,6 @@ alert:
 - `actions` - list of actions
 - `on-failure` - a conditionless action used in case of an alert failure
 
-### Steps
-```yaml
-steps:
-    - name: get-max-datetime
-      provider:
-      condition:
-```
-`Step` is built of:
-  - `name` - the step name (context will be accessible through `{{ steps.name.results }}`).
-  - `provider` - the data source.
-
 ### Provider
 ```yaml
 provider:
@@ -92,9 +81,9 @@ provider:
 - `compare_to` - whats `value` will be compared against
 - `compare_type` - all type-specific condition configuration
 
-### Actions
+### Steps/Actions
 ```yaml
-actions:
+steps/actions:
   - name: trigger-slack
     condition:
     - name: threshold-condition
@@ -109,7 +98,7 @@ actions:
         message: "DB datetime value ({{ actions.trigger-slack.conditions.threshold.0.compare_value }}) is greater than 1! 🚨"
 ```
 
-`Action` is built of:
+`Step/Action` is built of:
 - `name` - the name of the action.
 - `condition` - a list of conditions that
 - `provider` - the provider that will trigger the action.
diff --git a/keep/action/__init__.py b/keep/action/__init__.py
deleted file mode 100644
index e69de29bb2..0000000000
diff --git a/keep/action/action.py b/keep/action/action.py
deleted file mode 100644
index 7d58282732..0000000000
--- a/keep/action/action.py
+++ /dev/null
@@ -1,164 +0,0 @@
-import asyncio
-import inspect
-import logging
-from dataclasses import field
-
-from pydantic.dataclasses import dataclass
-
-from keep.conditions.condition_factory import ConditionFactory
-from keep.contextmanager.contextmanager import ContextManager
-from keep.exceptions.action_error import ActionError
-from keep.iohandler.iohandler import IOHandler
-from keep.providers.base.base_provider import BaseProvider
-from keep.throttles.throttle_factory import ThrottleFactory
-
-
-@dataclass(config={"arbitrary_types_allowed": True})
-class Action:
-    name: str
-    config: dict
-    provider: BaseProvider
-    provider_context: dict
-    conditions_results: dict = field(default_factory=dict)
-    conditions: list = field(default_factory=list)
-
-    def __post_init__(self):
-        self.logger = logging.getLogger(__name__)
-        self.io_handler = IOHandler()
-        self.context_manager = ContextManager.get_instance()
-        self.conditions = self.config.get("condition", [])
-
-    def run(self):
-        try:
-            if self.config.get("foreach"):
-                did_action_run = self._run_foreach()
-            else:
-                did_action_run = self._run_single()
-            return did_action_run
-        except Exception as e:
-            raise ActionError(e)
-
-    def _check_throttling(self, action_name):
-        throttling = self.config.get("throttle")
-        # if there is no throttling, return
-        if not throttling:
-            return False
-
-        throttling_type = throttling.get("type")
-        throttling_config = throttling.get("with")
-        throttle = ThrottleFactory.get_instance(throttling_type, throttling_config)
-        alert_id = self.context_manager.get_alert_id()
-        return throttle.check_throttling(action_name, alert_id)
-
-    def _run_foreach(self):
-        """Evaluate the action for each item, when using the `foreach` attribute (see foreach.md)"""
-        # the item holds the value we are going to iterate over
-        items = self.io_handler.render(self.config.get("foreach"))
-        any_action_run = False
-        # apply ALL conditions (the decision whether to run or not is made in the end)
-        for item in items:
-            self.context_manager.set_for_each_context(item)
-            did_action_run = self._run_single()
-            # If at least one item triggered an action, return True
-            # TODO - do it per item
-            if did_action_run:
-                any_action_run = True
-        return any_action_run
-
-    def _run_single(self):
-        # Initialize all conditions
-        conditions = []
-
-        for condition in self.conditions:
-            condition_name = condition.get("name", None)
-
-            if not condition_name:
-                raise Exception("Condition must have a name")
-
-            conditions.append(
-                ConditionFactory.get_condition(
-                    condition.get("type"),
-                    condition_name,
-                    condition,
-                )
-            )
-
-        for condition in conditions:
-            condition_compare_to = condition.get_compare_to()
-            condition_compare_value = condition.get_compare_value()
-            condition_result = condition.apply(
-                condition_compare_to, condition_compare_value
-            )
-            self.context_manager.set_condition_results(
-                self.name,
-                condition.condition_name,
-                condition.condition_type,
-                condition_compare_to,
-                condition_compare_value,
-                condition_result,
-                condition_alias=condition.condition_alias,
-                **condition.condition_context,
-            )
-
-        # Second, decide if need to run
-        # after all conditions are applied, check if we need to run
-        # there are 2 cases:
-        # 1. a "if" block is supplied, then use it
-        # 2. no "if" block is supplied, then use the AND between all conditions
-        if self.config.get("if"):
-            if_conf = self.config.get("if")
-        else:
-            # create a string of all conditions, separated by "and"
-            if_conf = " and ".join(
-                [f"{{{{ {condition.condition_alias} }}}} " for condition in conditions]
-            )
-
-        # Now check it
-        if if_conf:
-            if_met = self.io_handler.render(if_conf)
-            # Evaluate the condition string
-            if_met = eval(if_met)
-        else:
-            if_met = True
-
-        if not if_met:
-            self.logger.info(
-                "Action %s evaluated NOT to run, Reason: %s evaluated to false.",
-                self.config.get("name"),
-                if_conf,
-            )
-            return
-
-        self.logger.info("Action %s evaluated to run!", self.config.get("name"))
-
-        # Third, check throttling
-        # Now check if throttling is enabled
-        throttled = self._check_throttling(self.config.get("name"))
-        if throttled:
-            self.logger.info("Action %s is throttled", self.config.get("name"))
-            return
-
-        # Last, run the action
-        rendered_value = self.io_handler.render_context(self.provider_context)
-        # if the provider is async, run it in a new event loop
-        if inspect.iscoroutinefunction(self.provider.notify):
-            result = self._run_single_async()
-        # else, just run the provider
-        else:
-            self.provider.notify(**rendered_value)
-            return True
-
-    def _run_single_async(self):
-        """For async providers, run them in a new event loop
-
-        Raises:
-            ActionError: _description_
-        """
-        rendered_value = self.io_handler.render_context(self.provider_context)
-        # This is "magically solved" because of nest_asyncio but probably isn't best practice
-        loop = asyncio.new_event_loop()
-        task = loop.create_task(self.provider.notify(**rendered_value))
-        try:
-            loop.run_until_complete(task)
-        except Exception as e:
-            raise ActionError(e)
diff --git a/keep/alert/alert.py b/keep/alert/alert.py
index 438e0f63e4..952292522b 100644
--- a/keep/alert/alert.py
+++ b/keep/alert/alert.py
@@ -4,7 +4,6 @@
 
 from pydantic.dataclasses import dataclass
 
-from keep.action.action import Action
 from keep.contextmanager.contextmanager import ContextManager
 from keep.iohandler.iohandler import IOHandler
 from keep.step.step import Step, StepError
@@ -22,9 +21,9 @@ class Alert:
     alert_owners: typing.List[str]
     alert_tags: typing.List[str]
     alert_steps: typing.List[Step]
-    alert_actions: typing.List[Action]
+    alert_actions: typing.List[Step]
     alert_file: str = None
-    on_failure: Action = None
+    on_failure: Step = None
 
     def __post_init__(self):
         self.logger = logging.getLogger(__name__)
@@ -68,7 +67,7 @@ def run_steps(self):
                 raise
         self.logger.debug(f"Steps for alert {self.alert_id} ran successfully")
 
-    def run_action(self, action: Action):
+    def run_action(self, action: Step):
         self.logger.info("Running action %s", action.name)
         try:
             action_status = action.run()
diff --git a/keep/parser/parser.py b/keep/parser/parser.py
index 14f04ccd25..998a48a139 100644
--- a/keep/parser/parser.py
+++ b/keep/parser/parser.py
@@ -8,13 +8,12 @@
 import validators
 import yaml
 
-from keep.action.action import Action
 from keep.alert.alert import Alert
 from keep.contextmanager.contextmanager import ContextManager
 from keep.iohandler.iohandler import IOHandler
 from keep.providers.base.base_provider import BaseProvider
 from keep.providers.providers_factory import ProvidersFactory
-from keep.step.step import Step
+from keep.step.step import Step, StepType
 
 
 class Parser:
@@ -211,6 +210,7 @@ def _parse_steps(self, alert) -> typing.List[Step]:
                 step_config=_step,
                 provider=provider,
                 provider_parameters=provider_parameters,
+                step_type=StepType.STEP,
             )
             alerts_steps_parsed.append(step)
         self.logger.debug("Steps parsed successfully")
@@ -231,7 +231,7 @@ def _get_step_provider(self, _step: dict) -> dict:
         )
         return provider
 
-    def _get_action(self, action: dict, action_name: str | None = None) -> Action:
+    def _get_action(self, action: dict, action_name: str | None = None) -> Step:
         name = action_name or action.get("name")
         provider_config = action.get("provider").get("config")
         provider_context = action.get("provider").get("with", {})
@@ -242,25 +242,27 @@ def _get_action(self, action: dict, action_name: str | None = None) -> Action:
         provider = ProvidersFactory.get_provider(
             provider_id, provider_type, provider_config, **provider_context
         )
-        action = Action(
+        action = Step(
             name=name,
             provider=provider,
             config=action,
             provider_context=provider_context,
+            step_type=StepType.ACTION,
         )
         return action
 
-    def _parse_actions(self, alert) -> typing.List[Action]:
+    def _parse_actions(self, alert) -> typing.List[Step]:
         self.logger.debug("Parsing actions")
         alert_actions = alert.get("actions", [])
         alert_actions_parsed = []
         for _action in alert_actions:
             parsed_action = self._get_action(_action)
+            parsed_action.step_type = StepType.ACTION
             alert_actions_parsed.append(parsed_action)
         self.logger.debug("Actions parsed successfully")
         return alert_actions_parsed
 
-    def _get_on_failure_action(self, alert) -> Action | None:
+    def _get_on_failure_action(self, alert) -> Step | None:
         """
         Parse the on-failure action
 
diff --git a/keep/step/step.py b/keep/step/step.py
index 810675537a..7ed8a93042 100644
--- a/keep/step/step.py
+++ b/keep/step/step.py
@@ -1,25 +1,42 @@
+import asyncio
+from enum import Enum
+import inspect
 import logging
 from dataclasses import field
 
 import chevron
 from pydantic.dataclasses import dataclass
 
+from keep.conditions.condition_factory import ConditionFactory
 from keep.contextmanager.contextmanager import ContextManager
+from keep.exceptions.action_error import ActionError
 from keep.iohandler.iohandler import IOHandler
 from keep.providers.base.base_provider import BaseProvider
+from keep.throttles.throttle_factory import ThrottleFactory
+
+class StepType(Enum):
+    STEP = "step"
+    ACTION = "action"
 
 
 @dataclass(config={"arbitrary_types_allowed": True})
 class Step:
-    step_id: str
-    step_config: dict
-    provider: BaseProvider
-    provider_parameters: dict
+    name: str = field(default_factory=str)
+    config: dict = field(default_factory=dict)
+    step_id: str = field(default_factory=str)
+    step_config: dict = field(default_factory=dict)
+    provider: BaseProvider = field(default_factory=BaseProvider)
+    provider_context: dict = field(default_factory=dict)
+    provider_parameters: dict = field(default_factory=dict)
+    conditions_results: dict = field(default_factory=dict)
+    conditions: list = field(default_factory=list)
+    step_type: Enum = field(default=StepType.STEP)
 
     def __post_init__(self):
         self.io_handler = IOHandler()
         self.logger = logging.getLogger(__name__)
         self.context_manager = ContextManager.get_instance()
+        self.conditions = self.config.get("condition", [])
 
     @property
     def foreach(self):
@@ -27,25 +44,155 @@ def foreach(self):
 
     def run(self):
         try:
-            # Inject the context to the parameters
-            rendered_providers_parameters = {}
-            for parameter in self.provider_parameters:
-                rendered_providers_parameters[parameter] = self.io_handler.render(
-                    self.provider_parameters[parameter]
+            if self.config.get("foreach"):
+                did_action_run = self._run_foreach()
+            else:
+                did_action_run = self._run_single()
+            return did_action_run
+        except Exception as e:
+            raise ActionError(e)
+    
+    def _check_throttling(self, action_name):
+        throttling = self.config.get("throttle")
+        # if there is no throttling, return
+        if not throttling:
+            return False
+
+        throttling_type = throttling.get("type")
+        throttling_config = throttling.get("with")
+        throttle = ThrottleFactory.get_instance(throttling_type, throttling_config)
+        alert_id = self.context_manager.get_alert_id()
+        return throttle.check_throttling(action_name, alert_id)
+    
+    def _run_foreach(self):
+        """Evaluate the action for each item, when using the `foreach` attribute (see foreach.md)"""
+        # the item holds the value we are going to iterate over
+        items = self.io_handler.render(self.config.get("foreach"))
+        any_action_run = False
+        # apply ALL conditions (the decision whether to run or not is made in the end)
+        for item in items:
+            self.context_manager.set_for_each_context(item)
+            did_action_run = self._run_single()
+            # If at least one item triggered an action, return True
+            # TODO - do it per item
+            if did_action_run:
+                any_action_run = True
+        return any_action_run
+    
+    def _run_single(self):
+        # Initialize all conditions
+        conditions = []
+
+        for condition in self.conditions:
+            condition_name = condition.get("name", None)
+
+            if not condition_name:
+                raise Exception("Condition must have a name")
+
+            conditions.append(
+                ConditionFactory.get_condition(
+                    condition.get("type"),
+                    condition_name,
+                    condition,
                 )
-            step_output = self.provider.query(**rendered_providers_parameters)
-            # after the provider ran, let's update the context with the context of the provider
-            #                         so it'll be available for the alert.
-            extra_context = self.provider.expose()
-            rendered_providers_parameters.update(extra_context)
-            self.context_manager.set_step_provider_paremeters(
-                self.step_id, rendered_providers_parameters
             )
-        except Exception as e:
-            raise StepError(e)
 
-        return step_output
+        for condition in conditions:
+            condition_compare_to = condition.get_compare_to()
+            condition_compare_value = condition.get_compare_value()
+            condition_result = condition.apply(
+                condition_compare_to, condition_compare_value
+            )
+            self.context_manager.set_condition_results(
+                self.name,
+                condition.condition_name,
+                condition.condition_type,
+                condition_compare_to,
+                condition_compare_value,
+                condition_result,
+                condition_alias=condition.condition_alias,
+                **condition.condition_context,
+            )
+
+        # Second, decide if need to run
+        # after all conditions are applied, check if we need to run
+        # there are 2 cases:
+        # 1. a "if" block is supplied, then use it
+        # 2. no "if" block is supplied, then use the AND between all conditions
+        if self.config.get("if"):
+            if_conf = self.config.get("if")
+        else:
+            # create a string of all conditions, separated by "and"
+            if_conf = " and ".join(
+                [f"{{{{ {condition.condition_alias} }}}} " for condition in conditions]
+            )
+
+        # Now check it
+        if if_conf:
+            if_met = self.io_handler.render(if_conf)
+            # Evaluate the condition string
+            if_met = eval(if_met)
+        else:
+            if_met = True
+
+        if not if_met:
+            self.logger.info(
+                "Action %s evaluated NOT to run, Reason: %s evaluated to false.",
+                self.config.get("name"),
+                if_conf,
+            )
+            return
 
+        self.logger.info("Action %s evaluated to run!", self.config.get("name"))
 
+        # Third, check throttling
+        # Now check if throttling is enabled
+        throttled = self._check_throttling(self.config.get("name"))
+        if throttled:
+            self.logger.info("Action %s is throttled", self.config.get("name"))
+            return
+
+        # Last, run the action
+        rendered_value = self.io_handler.render_context(self.provider_context)
+        # if the provider is async, run it in a new event loop
+        if inspect.iscoroutinefunction(self.provider.notify):
+            result = self._run_single_async()
+        # else, just run the provider
+        else:
+            if self.step_type == StepType.STEP:
+                try:
+                    rendered_providers_parameters = {}
+                    for parameter in self.provider_parameters:
+                        rendered_providers_parameters[parameter] = self.io_handler.render(
+                            self.provider_parameters[parameter]
+                        )
+                    step_output =  self.provider.query(**rendered_value)
+                    extra_context = self.provider.expose()
+                    rendered_providers_parameters.update(extra_context)
+                    self.context_manager.set_step_provider_paremeters(
+                        self.step_id, rendered_providers_parameters
+                    )
+                except Exception as e:
+                    raise StepError(e)
+                
+                return step_output
+            self.provider.notify(**rendered_value)
+            return True
+        
+    def _run_single_async(self):
+        """For async providers, run them in a new event loop
+
+        Raises:
+            ActionError: _description_
+        """
+        rendered_value = self.io_handler.render_context(self.provider_context)
+        # This is "magically solved" because of nest_asyncio but probably isn't best practice
+        loop = asyncio.new_event_loop()
+        task = loop.create_task(self.provider.notify(**rendered_value))
+        try:
+            loop.run_until_complete(task)
+        except Exception as e:
+            raise ActionError(e)
+        
 class StepError(Exception):
     pass


Code review comments:
diff_hunk: @@ -1,51 +1,198 @@
+import asyncio
+from enum import Enum
+import inspect
 import logging
 from dataclasses import field
 
 import chevron
 from pydantic.dataclasses import dataclass
 
+from keep.conditions.condition_factory import ConditionFactory
 from keep.contextmanager.contextmanager import ContextManager
+from keep.exceptions.action_error import ActionError
 from keep.iohandler.iohandler import IOHandler
 from keep.providers.base.base_provider import BaseProvider
+from keep.throttles.throttle_factory import ThrottleFactory
+
+class StepType(Enum):
+    STEP = "step"
+    ACTION = "action"
 
 
 @dataclass(config={"arbitrary_types_allowed": True})
 class Step:
-    step_id: str
-    step_config: dict
-    provider: BaseProvider
-    provider_parameters: dict
+    name: str = field(default_factory=str)
+    config: dict = field(default_factory=dict)
+    step_id: str = field(default_factory=str)
+    step_config: dict = field(default_factory=dict)
+    provider: BaseProvider = field(default_factory=BaseProvider)
+    provider_context: dict = field(default_factory=dict)
+    provider_parameters: dict = field(default_factory=dict)
+    conditions_results: dict = field(default_factory=dict)
+    conditions: list = field(default_factory=list)
+    step_type: Enum = field(default=StepType.STEP)
 
     def __post_init__(self):
         self.io_handler = IOHandler()
         self.logger = logging.getLogger(__name__)
         self.context_manager = ContextManager.get_instance()
+        self.conditions = self.config.get("condition", [])
 
     @property
     def foreach(self):
         return self.step_config.get("foreach")
 
     def run(self):
         try:
-            # Inject the context to the parameters
-            rendered_providers_parameters = {}
-            for parameter in self.provider_parameters:
-                rendered_providers_parameters[parameter] = self.io_handler.render(
-                    self.provider_parameters[parameter]
+            if self.config.get("foreach"):
+                did_action_run = self._run_foreach()
+            else:
+                did_action_run = self._run_single()
+            return did_action_run
+        except Exception as e:
+            raise ActionError(e)
+    
+    def _check_throttling(self, action_name):
+        throttling = self.config.get("throttle")
+        # if there is no throttling, return
+        if not throttling:
+            return False
+
+        throttling_type = throttling.get("type")
+        throttling_config = throttling.get("with")
+        throttle = ThrottleFactory.get_instance(throttling_type, throttling_config)
+        alert_id = self.context_manager.get_alert_id()
+        return throttle.check_throttling(action_name, alert_id)
+    
+    def _run_foreach(self):
+        """Evaluate the action for each item, when using the `foreach` attribute (see foreach.md)"""
+        # the item holds the value we are going to iterate over
+        items = self.io_handler.render(self.config.get("foreach"))
+        any_action_run = False
+        # apply ALL conditions (the decision whether to run or not is made in the end)
+        for item in items:
+            self.context_manager.set_for_each_context(item)
+            did_action_run = self._run_single()
+            # If at least one item triggered an action, return True
+            # TODO - do it per item
+            if did_action_run:
+                any_action_run = True
+        return any_action_run
+    
+    def _run_single(self):
+        # Initialize all conditions
+        conditions = []
+
+        for condition in self.conditions:
+            condition_name = condition.get("name", None)
+
+            if not condition_name:
+                raise Exception("Condition must have a name")
+
+            conditions.append(
+                ConditionFactory.get_condition(
+                    condition.get("type"),
+                    condition_name,
+                    condition,
                 )
-            step_output = self.provider.query(**rendered_providers_parameters)
-            # after the provider ran, let's update the context with the context of the provider
-            #                         so it'll be available for the alert.
-            extra_context = self.provider.expose()
-            rendered_providers_parameters.update(extra_context)
-            self.context_manager.set_step_provider_paremeters(
-                self.step_id, rendered_providers_parameters
             )
-        except Exception as e:
-            raise StepError(e)
 
-        return step_output
+        for condition in conditions:
+            condition_compare_to = condition.get_compare_to()
+            condition_compare_value = condition.get_compare_value()
+            condition_result = condition.apply(
+                condition_compare_to, condition_compare_value
+            )
+            self.context_manager.set_condition_results(
+                self.name,
+                condition.condition_name,
+                condition.condition_type,
+                condition_compare_to,
+                condition_compare_value,
+                condition_result,
+                condition_alias=condition.condition_alias,
+                **condition.condition_context,
+            )
+
+        # Second, decide if need to run
+        # after all conditions are applied, check if we need to run
+        # there are 2 cases:
+        # 1. a "if" block is supplied, then use it
+        # 2. no "if" block is supplied, then use the AND between all conditions
+        if self.config.get("if"):
+            if_conf = self.config.get("if")
+        else:
+            # create a string of all conditions, separated by "and"
+            if_conf = " and ".join(
+                [f"{{{{ {condition.condition_alias} }}}} " for condition in conditions]
+            )
+
+        # Now check it
+        if if_conf:
+            if_met = self.io_handler.render(if_conf)
+            # Evaluate the condition string
+            if_met = eval(if_met)
+        else:
+            if_met = True
+
+        if not if_met:
+            self.logger.info(
+                "Action %s evaluated NOT to run, Reason: %s evaluated to false.",
+                self.config.get("name"),
+                if_conf,
+            )
+            return
 
+        self.logger.info("Action %s evaluated to run!", self.config.get("name"))
 
+        # Third, check throttling
+        # Now check if throttling is enabled
+        throttled = self._check_throttling(self.config.get("name"))
+        if throttled:
+            self.logger.info("Action %s is throttled", self.config.get("name"))
+            return
+
+        # Last, run the action
+        rendered_value = self.io_handler.render_context(self.provider_context)
+        # if the provider is async, run it in a new event loop
+        if inspect.iscoroutinefunction(self.provider.notify):
+            result = self._run_single_async()
+        # else, just run the provider
+        else:
+            if self.step_type == StepType.STEP:
+                try:
+                    rendered_providers_parameters = {}
+                    for parameter in self.provider_parameters:
+                        rendered_providers_parameters[parameter] = self.io_handler.render(
+                            self.provider_parameters[parameter]
+                        )
+                    step_output =  self.provider.query(**rendered_value)
+                    extra_context = self.provider.expose()
+                    rendered_providers_parameters.update(extra_context)
+                    self.context_manager.set_step_provider_paremeters(
+                        self.step_id, rendered_providers_parameters
+                    )
+                except Exception as e:
+                    raise StepError(e)
+                
+                return step_output
+            self.provider.notify(**rendered_value)
+            return True
path: keep/step/step.py
text: @user1:
I think we need to try and keep logic as similar as possible, wdyt about the following:

```suggestion
        if inspect.iscoroutinefunction(self.provider.notify) or inspect.iscoroutinefunction(self.provider.query):
            result = self._run_single_async()
        # else, just run the provider
        else:
              try:
                  rendered_providers_parameters = {}
                  for parameter in self.provider_parameters:
                      rendered_providers_parameters[parameter] = self.io_handler.render(
                          self.provider_parameters[parameter]
                      )
                  step_output =  self.provider.query(**rendered_value) if self.step_type == StepType.STEP else self.provider.notify(**rendered_value)
                  extra_context = self.provider.expose()
                  rendered_providers_parameters.update(extra_context)
                  self.context_manager.set_step_provider_paremeters(
                      self.step_id, rendered_providers_parameters
                  )
              except Exception as e:
                  raise StepError(e)
```

@author:
Hi @user1, Yes that is achievable. I can refactor it to look like this;

```
if inspect.iscoroutinefunction(self.provider.notify):
    result = self._run_single_async()
# else, just run the provider
else:
    try:
        rendered_providers_parameters = {}
        for parameter in self.provider_parameters:
            rendered_providers_parameters[parameter] = self.io_handler.render(
                self.provider_parameters[parameter]
            )

        if self.step_type == StepType.STEP:
            self.provider.query(**rendered_value)
        else:
            self.provider.notify(**rendered_value)
            
        extra_context = self.provider.expose()
        rendered_providers_parameters.update(extra_context)
        self.context_manager.set_step_provider_paremeters(
            self.step_id, rendered_providers_parameters
        )
    except Exception as e:
        raise StepError(e)
    
    return True
```

The only thing is that we would need to take out the `step_output`.

What do you think?

diff_hunk: @@ -242,25 +242,27 @@ def _get_action(self, action: dict, action_name: str | None = None) -> Action:
         provider = ProvidersFactory.get_provider(
             provider_id, provider_type, provider_config, **provider_context
         )
-        action = Action(
+        action = Step(
             name=name,
             provider=provider,
             config=action,
             provider_context=provider_context,
+            step_type=StepType.ACTION,
         )
         return action
 
-    def _parse_actions(self, alert) -> typing.List[Action]:
+    def _parse_actions(self, alert) -> typing.List[Step]:
         self.logger.debug("Parsing actions")
         alert_actions = alert.get("actions", [])
         alert_actions_parsed = []
         for _action in alert_actions:
             parsed_action = self._get_action(_action)
+            parsed_action.step_type = StepType.ACTION
path: keep/parser/parser.py
text: @user1:
Seems redundant, as https://github.com/keephq/keep/pull/186/files#diff-d70b456d958384fcd43980d14f1c8335a30d72522715c7d9b6b7a9d02845a12dR245 already handles that, no?

@author:
Yes. That has been handled already in the `_get_action` method. Thank you for pointing that out.

diff_hunk: @@ -1,51 +1,198 @@
+import asyncio
+from enum import Enum
+import inspect
 import logging
 from dataclasses import field
 
 import chevron
 from pydantic.dataclasses import dataclass
 
+from keep.conditions.condition_factory import ConditionFactory
 from keep.contextmanager.contextmanager import ContextManager
+from keep.exceptions.action_error import ActionError
 from keep.iohandler.iohandler import IOHandler
 from keep.providers.base.base_provider import BaseProvider
+from keep.throttles.throttle_factory import ThrottleFactory
+
+class StepType(Enum):
+    STEP = "step"
+    ACTION = "action"
 
 
 @dataclass(config={"arbitrary_types_allowed": True})
 class Step:
-    step_id: str
-    step_config: dict
-    provider: BaseProvider
-    provider_parameters: dict
+    name: str = field(default_factory=str)
+    config: dict = field(default_factory=dict)
+    step_id: str = field(default_factory=str)
+    step_config: dict = field(default_factory=dict)
+    provider: BaseProvider = field(default_factory=BaseProvider)
+    provider_context: dict = field(default_factory=dict)
+    provider_parameters: dict = field(default_factory=dict)
+    conditions_results: dict = field(default_factory=dict)
+    conditions: list = field(default_factory=list)
+    step_type: Enum = field(default=StepType.STEP)
 
     def __post_init__(self):
         self.io_handler = IOHandler()
         self.logger = logging.getLogger(__name__)
         self.context_manager = ContextManager.get_instance()
+        self.conditions = self.config.get("condition", [])
 
     @property
     def foreach(self):
         return self.step_config.get("foreach")
 
     def run(self):
         try:
-            # Inject the context to the parameters
-            rendered_providers_parameters = {}
-            for parameter in self.provider_parameters:
-                rendered_providers_parameters[parameter] = self.io_handler.render(
-                    self.provider_parameters[parameter]
+            if self.config.get("foreach"):
+                did_action_run = self._run_foreach()
+            else:
+                did_action_run = self._run_single()
+            return did_action_run
+        except Exception as e:
+            raise ActionError(e)
+    
+    def _check_throttling(self, action_name):
+        throttling = self.config.get("throttle")
+        # if there is no throttling, return
+        if not throttling:
+            return False
+
+        throttling_type = throttling.get("type")
+        throttling_config = throttling.get("with")
+        throttle = ThrottleFactory.get_instance(throttling_type, throttling_config)
+        alert_id = self.context_manager.get_alert_id()
+        return throttle.check_throttling(action_name, alert_id)
+    
+    def _run_foreach(self):
+        """Evaluate the action for each item, when using the `foreach` attribute (see foreach.md)"""
+        # the item holds the value we are going to iterate over
+        items = self.io_handler.render(self.config.get("foreach"))
+        any_action_run = False
+        # apply ALL conditions (the decision whether to run or not is made in the end)
+        for item in items:
+            self.context_manager.set_for_each_context(item)
+            did_action_run = self._run_single()
+            # If at least one item triggered an action, return True
+            # TODO - do it per item
+            if did_action_run:
+                any_action_run = True
+        return any_action_run
+    
+    def _run_single(self):
+        # Initialize all conditions
+        conditions = []
+
+        for condition in self.conditions:
+            condition_name = condition.get("name", None)
+
+            if not condition_name:
+                raise Exception("Condition must have a name")
+
+            conditions.append(
+                ConditionFactory.get_condition(
+                    condition.get("type"),
+                    condition_name,
+                    condition,
                 )
-            step_output = self.provider.query(**rendered_providers_parameters)
-            # after the provider ran, let's update the context with the context of the provider
-            #                         so it'll be available for the alert.
-            extra_context = self.provider.expose()
-            rendered_providers_parameters.update(extra_context)
-            self.context_manager.set_step_provider_paremeters(
-                self.step_id, rendered_providers_parameters
             )
-        except Exception as e:
-            raise StepError(e)
 
-        return step_output
+        for condition in conditions:
+            condition_compare_to = condition.get_compare_to()
+            condition_compare_value = condition.get_compare_value()
+            condition_result = condition.apply(
+                condition_compare_to, condition_compare_value
+            )
+            self.context_manager.set_condition_results(
+                self.name,
+                condition.condition_name,
+                condition.condition_type,
+                condition_compare_to,
+                condition_compare_value,
+                condition_result,
+                condition_alias=condition.condition_alias,
+                **condition.condition_context,
+            )
+
+        # Second, decide if need to run
+        # after all conditions are applied, check if we need to run
+        # there are 2 cases:
+        # 1. a "if" block is supplied, then use it
+        # 2. no "if" block is supplied, then use the AND between all conditions
+        if self.config.get("if"):
+            if_conf = self.config.get("if")
+        else:
+            # create a string of all conditions, separated by "and"
+            if_conf = " and ".join(
+                [f"{{{{ {condition.condition_alias} }}}} " for condition in conditions]
+            )
+
+        # Now check it
+        if if_conf:
+            if_met = self.io_handler.render(if_conf)
+            # Evaluate the condition string
+            if_met = eval(if_met)
+        else:
+            if_met = True
+
+        if not if_met:
+            self.logger.info(
+                "Action %s evaluated NOT to run, Reason: %s evaluated to false.",
+                self.config.get("name"),
+                if_conf,
+            )
+            return
 
+        self.logger.info("Action %s evaluated to run!", self.config.get("name"))
 
+        # Third, check throttling
+        # Now check if throttling is enabled
+        throttled = self._check_throttling(self.config.get("name"))
+        if throttled:
+            self.logger.info("Action %s is throttled", self.config.get("name"))
+            return
+
+        # Last, run the action
+        rendered_value = self.io_handler.render_context(self.provider_context)
+        # if the provider is async, run it in a new event loop
+        if inspect.iscoroutinefunction(self.provider.notify):
+            result = self._run_single_async()
+        # else, just run the provider
+        else:
+            if self.step_type == StepType.STEP:
+                try:
+                    rendered_providers_parameters = {}
+                    for parameter in self.provider_parameters:
+                        rendered_providers_parameters[parameter] = self.io_handler.render(
+                            self.provider_parameters[parameter]
+                        )
+                    step_output =  self.provider.query(**rendered_value)
+                    extra_context = self.provider.expose()
+                    rendered_providers_parameters.update(extra_context)
+                    self.context_manager.set_step_provider_paremeters(
+                        self.step_id, rendered_providers_parameters
+                    )
+                except Exception as e:
+                    raise StepError(e)
+                
+                return step_output
+            self.provider.notify(**rendered_value)
+            return True
+        
+    def _run_single_async(self):
+        """For async providers, run them in a new event loop
+
+        Raises:
+            ActionError: _description_
+        """
+        rendered_value = self.io_handler.render_context(self.provider_context)
+        # This is "magically solved" because of nest_asyncio but probably isn't best practice
+        loop = asyncio.new_event_loop()
+        task = loop.create_task(self.provider.notify(**rendered_value))
path: keep/step/step.py
text: @user1:
```suggestion
        if self.step_type == StepType.STEP:
            task = loop.create_task(self.provider.query(**rendered_value))
        else:
            task = loop.create_task(self.provider.notify(**rendered_value))
```

@author:
Ok. I will add that. Thank you for pointing that out.