Process as subprocess conditions

.pre-commit-config.yaml

@@ -18,7 +18,7 @@ repos:
       - id: requirements-txt-fixer
 
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.8.0
+    rev: v0.8.6
     hooks:
       - id: ruff-format
       - id: ruff
@@ -32,7 +32,7 @@ repos:
         additional_dependencies: ["flake8-pyproject", "wemake-python-styleguide"]
 
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: 'v1.13.0'
+    rev: 'v1.14.1'
     hooks:
       - id: mypy
         additional_dependencies: [ "no_implicit_optional" ]

README.md

@@ -18,9 +18,13 @@ pip install logic-processes-layer
 
 ## New Features
 
-- ProcessAsSubprocess: Use any process as a subprocess.
-- InitMapper: Simplifies process initialization with attribute mapping from the context.
-- ProcessAttr: Retrieve attributes from the process context or directly from the process.
+- **ProcessAsSubprocess**: Use any process as a subprocess.
+- **InitMapper**: Simplifies process initialization with attribute mapping from the context.
+- **ProcessAttr**: Retrieve attributes from the process context or directly from the process.
+- **Conditions Support**: Add logical conditions to control the execution of processes.
+  - **AttrCondition**: Define conditions based on attributes of the process or context.
+  - **FunctionCondition**: Wrap custom functions as conditions.
+  - **Logical Operators**: Combine conditions with `&` (AND), `|` (OR), `~` (NOT), and `^` (XOR) for advanced logic.
 - [Examples](tests/examples) of how to use the logic_processes_layer package.
 
 

logic_processes_layer/extensions/conditions/__init__.py

@@ -0,0 +1,5 @@
+from __future__ import annotations
+
+from .condition import *
+from .operator_enums import *
+from .skiped import *

logic_processes_layer/extensions/conditions/condition.py

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+
+__all__ = ("AttrCondition", "FunctionCondition", "OperatorCondition")
+
+from functools import partial, reduce
+import typing
+
+from ...context import BaseProcessorContext
+from .operator_enums import OperatorEnum
+
+
+if typing.TYPE_CHECKING:
+    from ...protocols import CallableConditionProtocol
+    from ..mappers import ProcessAttr
+
+ContextT = typing.TypeVar("ContextT", bound=BaseProcessorContext)
+OperatorCallablesT = typing.Callable[[typing.Iterable[typing.Any]], bool]
+OperatorMapT = typing.Dict[OperatorEnum, OperatorCallablesT]
+
+
+class OperatorCondition(typing.Generic[ContextT]):
+    operator_map: typing.ClassVar[OperatorMapT] = {
+        OperatorEnum.AND: all,
+        OperatorEnum.OR: any,
+        OperatorEnum.XOR: partial(reduce, lambda itm, other: itm ^ other),
+    }
+
+    def __init__(
+        self,
+        conditions: typing.Iterable[CallableConditionProtocol],
+        *,
+        operator: OperatorEnum = OperatorEnum.AND,
+        negated: bool = False,
+    ):
+        self.conditions = conditions
+        self.negated = negated
+        self.operator = operator
+
+    def __call__(self, context: ContextT) -> bool:
+        operator_f = self.operator_map[self.operator]
+        result = operator_f(bool(condition(context)) for condition in self.conditions)
+        return not result if self.negated else result
+
+    def __invert__(self) -> OperatorCondition:
+        return OperatorCondition([self], operator=self.operator, negated=not self.negated)
+
+    def __and__(self, other: CallableConditionProtocol) -> OperatorCondition:
+        return OperatorCondition([self, other], operator=OperatorEnum.AND)
+
+    def __or__(self, other: CallableConditionProtocol) -> OperatorCondition:
+        return OperatorCondition([self, other], operator=OperatorEnum.OR)
+
+    def __xor__(self, other: CallableConditionProtocol) -> OperatorCondition:
+        return OperatorCondition([self, other], operator=OperatorEnum.XOR)
+
+
+class AttrCondition(OperatorCondition[ContextT]):
+    def __init__(self, process_attr: ProcessAttr, *, negated: bool = False):
+        self.process_attr = process_attr
+        super().__init__(operator=OperatorEnum.AND, conditions=[self], negated=negated)
+
+    def __call__(self, context: ContextT) -> bool:
+        value = self.process_attr.get_value(context)
+        result = bool(value)
+        return not result if self.negated else result
+
+
+class FunctionCondition(OperatorCondition[ContextT]):
+    def __init__(self, func: CallableConditionProtocol, *, negated: bool = False):
+        super().__init__(operator=OperatorEnum.AND, conditions=[func], negated=negated)

logic_processes_layer/extensions/conditions/operator_enums.py

@@ -0,0 +1,9 @@
+from __future__ import annotations
+
+import enum
+
+
+class OperatorEnum(enum.Enum):
+    AND = "AND"
+    OR = "OR"
+    XOR = "XOR"

logic_processes_layer/extensions/conditions/skiped.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+
+__all__ = ("ConditionSkipped",)
+
+import dataclasses
+import typing
+
+from ...processors import BaseProcessor
+
+
+ProcessT = typing.TypeVar("ProcessT", bound=BaseProcessor)
+
+
+@dataclasses.dataclass(unsafe_hash=True)
+class ConditionSkipped(typing.Generic[ProcessT]):
+    process: ProcessT
+
+    def __str__(self):
+        return f"Conditions skipped for {self.process}"

logic_processes_layer/extensions/process_as_subprocess.py

@@ -7,7 +7,8 @@
 import typing
 
 from ..processors import BaseProcessor
-from ..sub_processors import BaseSubprocessor
+from ..sub_processors import BaseSubprocessor, ContextT
+from .conditions import ConditionSkipped
 
 
 if typing.TYPE_CHECKING:
@@ -19,11 +20,21 @@
 @dataclasses.dataclass(unsafe_hash=True)
 class ProcessAsSubprocess(BaseSubprocessor, typing.Generic[ProcessT]):
     process_cls: type[ProcessT]
-    init_mapper: None | InitMapper = dataclasses.field(hash=False, default=None)
+    init_mapper: InitMapper | None = dataclasses.field(hash=False, default=None)
+    conditions: typing.Iterable[typing.Callable[[ContextT], bool]] = dataclasses.field(
+        hash=False, default_factory=tuple
+    )
 
     def __call__(self):
         args = ()
         kwargs: dict[str, typing.Any] = {}
         if self.init_mapper is not None:
             args, kwargs = self.init_mapper(self.context)
-        return self.process_cls(*args, **kwargs)()
+
+        subprocessor = self.process_cls(*args, **kwargs)
+        if not self.check_conditions():
+            return ConditionSkipped(subprocessor)
+        return subprocessor()
+
+    def check_conditions(self) -> bool:
+        return all(condition(self.context) for condition in self.conditions)

logic_processes_layer/protocols/__init__.py

@@ -0,0 +1,3 @@
+from __future__ import annotations
+
+from .condition import *

logic_processes_layer/protocols/condition.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+
+__all__ = ("CallableConditionProtocol",)
+
+import typing
+
+from ..context import BaseProcessorContext
+
+
+ContextT_contra = typing.TypeVar("ContextT_contra", bound=BaseProcessorContext, contravariant=True)
+
+
+@typing.runtime_checkable
+class CallableConditionProtocol(typing.Protocol[ContextT_contra]):
+    def __call__(self, context: ContextT_contra) -> bool: ...  # noqa: WPS220, WPS428

logic_processes_layer/sub_processors/base.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 
-__all__ = ("BaseSubprocessor",)
+__all__ = ("BaseSubprocessor", "CallResultT", "ContextT")
 
 import dataclasses
 import typing

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "logic_processes_layer"
-version = "1.1.4"
+version = "1.2025.01.06"
 requires-python = ">=3.8"
 description = "Abstractions for create business logic"
 readme = "README.md"
@@ -18,6 +18,7 @@ classifiers = ["License :: OSI Approved :: MIT License"]
 max-line-length = 120
 inline_quotes = "double"
 format = "wemake"
+select = ["WPS"]
 
 ignore = [
     "C", "D", "F", "DAR", "RST", "Q", "I", "S",
@@ -28,4 +29,6 @@ ignore = [
     "WPS602", "WPS605"]
 
 per-file-ignores = [
-    "__init__.py:WPS347, WPS440",]
+    "__init__.py:WPS347, WPS440",
+    "*_enums.py:WPS115",
+    "tests/test_*.py:WPS202, WPS442",]

ruff.toml

@@ -16,7 +16,7 @@ ignore = [
     "TD001", "TID252",
 ]
 
-external = ["WPS221", "WPS432", "WPS529", "WPS601","WPS428"]
+external = ["WPS220", "WPS221", "WPS432", "WPS529", "WPS601","WPS428"]
 
 [lint.per-file-ignores]
 "__init__.py" = ["F401", "F403"]

tests/examples/processors/process_with_conditions.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+import dataclasses
+
+from logic_processes_layer import BaseProcessor
+from logic_processes_layer.extensions import InitMapper, ProcessAsSubprocess, ProcessAttr
+from logic_processes_layer.extensions.conditions import AttrCondition, FunctionCondition
+
+
+_ONE_H = 100
+
+
+@dataclasses.dataclass
+class NotifyClientProcess(BaseProcessor):
+    notification_message: str
+
+    def run(self): ...  # noqa: WPS428
+
+
+_order_is_completed_condition: AttrCondition = AttrCondition(ProcessAttr("is_completed"))
+_order_amount_condition: FunctionCondition = FunctionCondition(lambda context: context.process.order_amount > _ONE_H)
+_client_agreed_condition: AttrCondition = AttrCondition(ProcessAttr("client_agreed_to_notifications"))
+_combined_condition = _order_is_completed_condition & (_order_amount_condition | _client_agreed_condition)
+
+_notify_client_process = ProcessAsSubprocess(
+    process_cls=NotifyClientProcess,
+    init_mapper=InitMapper(notification_message=ProcessAttr("notification_message")),
+    conditions=(_combined_condition,),
+)
+
+
+@dataclasses.dataclass
+class OrderStatusProcess(BaseProcessor):
+    is_completed: bool
+    client_agreed_to_notifications: bool
+    order_amount: float
+    notification_message: str = dataclasses.field(init=False, default="Order completed successfully!")
+
+    post_run = (_notify_client_process,)
+
+    def run(self): ...  # noqa: WPS428

tests/test_conditions.py

@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+from unittest.mock import Mock
+
+import pytest
+
+from logic_processes_layer.extensions import ProcessAttr
+from logic_processes_layer.extensions.conditions import AttrCondition, FunctionCondition
+
+
+_TWO = 2
+
+
+@pytest.fixture
+def mock_process_attr():
+    return Mock(spec=ProcessAttr)
+
+
+@pytest.fixture
+def mock_context():
+    return Mock()
+
+
+@pytest.fixture
+def attr_condition(mock_process_attr):
+    return AttrCondition(mock_process_attr)
+
+
+def test_attr_condition_positive(attr_condition, mock_process_attr, mock_context):
+    mock_process_attr.get_value.return_value = True
+    assert attr_condition(mock_context) is True
+    mock_process_attr.get_value.assert_called_once_with(mock_context)
+
+
+def test_attr_condition_negative(attr_condition, mock_process_attr, mock_context):
+    mock_process_attr.get_value.return_value = False
+    assert attr_condition(mock_context) is False
+    mock_process_attr.get_value.assert_called_once_with(mock_context)
+
+
+def test_attr_condition_negated(attr_condition, mock_process_attr, mock_context):
+    mock_process_attr.get_value.return_value = True
+    condition = ~attr_condition
+
+    assert condition(mock_context) is False
+    mock_process_attr.get_value.assert_called_once_with(mock_context)
+
+
+def test_function_condition_positive(mock_context):
+    mock_func = Mock(return_value=True)
+    condition: FunctionCondition = FunctionCondition(mock_func)
+
+    assert condition(mock_context) is True
+    mock_func.assert_called_once_with(mock_context)
+
+
+def test_function_condition_negative(mock_context):
+    mock_func = Mock(return_value=False)
+    condition: FunctionCondition = FunctionCondition(mock_func)
+
+    assert condition(mock_context) is False
+    mock_func.assert_called_once_with(mock_context)
+
+
+def test_function_condition_negated(mock_context):
+    mock_func = Mock(return_value=True)
+    condition = ~FunctionCondition(mock_func)
+
+    assert condition(mock_context) is False
+    mock_func.assert_called_once_with(mock_context)
+
+
+def test_operator_condition_and(attr_condition, mock_process_attr, mock_context):
+    mock_process_attr.get_value.side_effect = (True, True)
+    condition1 = attr_condition
+    condition2 = attr_condition
+    combined_condition = condition1 & condition2
+
+    assert combined_condition(mock_context) is True
+    assert mock_process_attr.get_value.call_count == _TWO
+
+
+def test_operator_condition_or(attr_condition, mock_process_attr, mock_context):
+    mock_process_attr.get_value.side_effect = (False, True)
+    condition1 = attr_condition
+    condition2 = attr_condition
+    combined_condition = condition1 | condition2
+
+    assert combined_condition(mock_context) is True
+    assert mock_process_attr.get_value.call_count == _TWO
+
+
+def test_operator_condition_complex(attr_condition, mock_process_attr, mock_context):
+    mock_process_attr.get_value.side_effect = (True, False, True)
+    condition1 = attr_condition
+    condition2 = ~attr_condition
+    condition3 = attr_condition
+    condition4: FunctionCondition = FunctionCondition(lambda _: True)
+    combined_condition = condition1 & (condition2 | condition3) & condition4
+
+    assert combined_condition(mock_context) is True
+    assert mock_process_attr.get_value.call_count == _TWO