Process as subprocess conditions (#26)

* update: pre-commit hooks
* update: flake8 selected
* update: versioning format
* add: conditions support
* Process as subprocess conditions (tests + doc) (#27)

* add: condition tests
update: pyproject.toml
del: attrs from requirements_dev.txt
* add: example with conditions
add: test for the process with conditions
* update: doc

---------

Co-authored-by: Sergei Aleshin <sergei.aleshin@alludo.com>

---------

Co-authored-by: GefMar <sergei.romanchuk@alludo.com>
Co-authored-by: Sergei Aleshin <66841202+Nishela@users.noreply.github.com>
Co-authored-by: Sergei Aleshin <sergei.aleshin@alludo.com>

Author: 4 people

.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