Merge pull request #150 from rsinger86/feature/generalize-conditions

Make conditions more flexible by extracting them from decorator signature to allow arbitrary conditions using callables

Author: EnriqueSoria

django_lifecycle/conditions/__init__.py

@@ -0,0 +1,121 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+from typing import Iterable
+from typing import Union
+
+from django_lifecycle import NotSet
+from django_lifecycle.conditions.base import ChainableCondition
+
+
+__all__ = [
+    "WhenFieldValueWas",
+    "WhenFieldValueIs",
+    "WhenFieldHasChanged",
+    "WhenFieldValueIsNot",
+    "WhenFieldValueWasNot",
+    "WhenFieldValueChangesTo",
+    "Always",
+]
+
+
+@dataclass
+class WhenFieldValueWas(ChainableCondition):
+    field_name: str
+    value: Any = "*"
+
+    def __call__(
+        self, instance: Any, update_fields: Union[Iterable[str], None] = None
+    ) -> bool:
+        return self.value in (instance.initial_value(self.field_name), "*")
+
+
+@dataclass
+class WhenFieldValueIs(ChainableCondition):
+    field_name: str
+    value: Any = "*"
+
+    def __call__(
+        self, instance: Any, update_fields: Union[Iterable[str], None] = None
+    ) -> bool:
+        return self.value in (instance._current_value(self.field_name), "*")
+
+
+@dataclass
+class WhenFieldHasChanged(ChainableCondition):
+    field_name: str
+    has_changed: bool | None = None
+
+    def __call__(
+        self, instance: Any, update_fields: Union[Iterable[str], None] = None
+    ) -> bool:
+        is_partial_fields_update = update_fields is not None
+        is_synced = (
+            is_partial_fields_update is False or self.field_name in update_fields
+        )
+        if not is_synced:
+            return False
+
+        return self.has_changed is None or self.has_changed == instance.has_changed(
+            self.field_name
+        )
+
+
+@dataclass
+class WhenFieldValueIsNot(ChainableCondition):
+    field_name: str
+    value: Any = NotSet
+
+    def __call__(
+        self, instance: Any, update_fields: Union[Iterable[str], None] = None
+    ) -> bool:
+        return (
+            self.value is NotSet
+            or instance._current_value(self.field_name) != self.value
+        )
+
+
+@dataclass
+class WhenFieldValueWasNot(ChainableCondition):
+    field_name: str
+    value: Any = NotSet
+
+    def __call__(
+        self, instance: Any, update_fields: Union[Iterable[str], None] = None
+    ) -> bool:
+        return (
+            self.value is NotSet
+            or instance.initial_value(self.field_name) != self.value
+        )
+
+
+@dataclass
+class WhenFieldValueChangesTo(ChainableCondition):
+    field_name: str
+    value: Any = NotSet
+
+    def __call__(
+        self, instance: Any, update_fields: Union[Iterable[str], None] = None
+    ) -> bool:
+        is_partial_fields_update = update_fields is not None
+        is_synced = (
+            is_partial_fields_update is False or self.field_name in update_fields
+        )
+        if not is_synced:
+            return False
+
+        value_has_changed = bool(
+            instance.initial_value(self.field_name) != self.value
+        )
+        new_value_is_the_expected = bool(
+            instance._current_value(self.field_name) == self.value
+        )
+        return self.value is NotSet or (
+            value_has_changed and new_value_is_the_expected
+        )
+
+
+class Always:
+    def __call__(self, instance: Any, update_fields=None):
+        return True

django_lifecycle/conditions/base.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+import operator
+from dataclasses import dataclass
+from typing import Any
+from typing import Callable
+from typing import Iterable
+from typing import Union
+
+from django_lifecycle import types
+
+
+@dataclass
+class ChainedCondition:
+    def __init__(
+        self,
+        left: types.Condition,
+        right: types.Condition,
+        operator: Callable[[Any, Any], bool],
+    ):
+        self.left = left
+        self.right = right
+        self.operator = operator
+
+    def __and__(self, other):
+        return ChainedCondition(self, other, operator=operator.and_)
+
+    def __or__(self, other):
+        return ChainedCondition(self, other, operator=operator.or_)
+
+    def __call__(
+        self, instance: Any, update_fields: Union[Iterable[str], None] = None
+    ) -> bool:
+        left_result = self.left(instance, update_fields)
+        right_result = self.right(instance, update_fields)
+        return self.operator(left_result, right_result)
+
+
+class ChainableCondition:
+    """Base class for defining chainable conditions using `&` and `|`"""
+
+    def __and__(self, other) -> ChainedCondition:
+        return ChainedCondition(self, other, operator=operator.and_)
+
+    def __or__(self, other) -> ChainedCondition:
+        return ChainedCondition(self, other, operator=operator.or_)
+
+    def __call__(
+        self, instance: Any, update_fields: Union[Iterable[str], None] = None
+    ) -> bool:
+        ...

django_lifecycle/conditions/legacy.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+from typing import List
+from typing import Optional
+
+from django_lifecycle import NotSet
+from django_lifecycle.conditions.base import ChainableCondition
+from django_lifecycle.conditions import WhenFieldValueChangesTo
+from django_lifecycle.conditions import WhenFieldHasChanged
+from django_lifecycle.conditions import WhenFieldValueIsNot
+from django_lifecycle.conditions import WhenFieldValueIs
+from django_lifecycle.conditions import WhenFieldValueWas
+from django_lifecycle.conditions import WhenFieldValueWasNot
+
+
+@dataclass
+class When(ChainableCondition):
+    when: Optional[str] = None
+    was: Any = "*"
+    is_now: Any = "*"
+    has_changed: Optional[bool] = None
+    is_not: Any = NotSet
+    was_not: Any = NotSet
+    changes_to: Any = NotSet
+
+    def __call__(self, instance: Any, update_fields=None) -> bool:
+        has_changed_condition = WhenFieldHasChanged(
+            self.when, has_changed=self.has_changed
+        )
+        if not has_changed_condition(instance, update_fields=update_fields):
+            return False
+
+        changes_to_condition = WhenFieldValueChangesTo(self.when, value=self.changes_to)
+        if not changes_to_condition(instance, self.when):
+            return False
+
+        is_now_condition = WhenFieldValueIs(self.when, value=self.is_now)
+        if not is_now_condition(instance, self.when):
+            return False
+
+        was_condition = WhenFieldValueWas(self.when, value=self.was)
+        if not was_condition(instance, self.when):
+            return False
+
+        was_not_condition = WhenFieldValueWasNot(self.when, value=self.was_not)
+        if not was_not_condition(instance, self.when):
+            return False
+
+        is_not_condition = WhenFieldValueIsNot(self.when, value=self.is_not)
+        if not is_not_condition(instance, self.when):
+            return False
+
+        return True
+
+
+@dataclass
+class WhenAny:
+    when_any: Optional[List[str]] = None
+    was: Any = "*"
+    is_now: Any = "*"
+    has_changed: Optional[bool] = None
+    is_not: Any = NotSet
+    was_not: Any = NotSet
+    changes_to: Any = NotSet
+
+    def __call__(self, instance: Any, update_fields=None) -> bool:
+        conditions = (
+            When(
+                when=field,
+                was=self.was,
+                is_now=self.is_now,
+                has_changed=self.has_changed,
+                is_not=self.is_not,
+                was_not=self.was_not,
+                changes_to=self.changes_to,
+            )
+            for field in self.when_any
+        )
+        return any(
+            condition(instance, update_fields=update_fields) for condition in conditions
+        )

django_lifecycle/decorators.py

@@ -1,9 +1,17 @@
+from __future__ import annotations
+
+import operator
+from django_lifecycle import types
 from dataclasses import dataclass
+from functools import reduce
 from functools import wraps
 from typing import Any
+from typing import Callable
 from typing import List, Optional
 
 from django_lifecycle import NotSet
+from .conditions import Always
+from .conditions.legacy import When
 from .dataclass_validation import Validations
 from .hooks import VALID_HOOKS
 from .priority import DEFAULT_PRIORITY
@@ -16,6 +24,11 @@ class DjangoLifeCycleException(Exception):
 @dataclass(order=False)
 class HookConfig(Validations):
     hook: str
+    on_commit: bool = False
+    priority: int = DEFAULT_PRIORITY
+    condition: Optional[types.Condition] = None
+
+    # Legacy parameters
     when: Optional[str] = None
     when_any: Optional[List[str]] = None
     was: Any = "*"
@@ -24,8 +37,57 @@ class HookConfig(Validations):
     is_not: Any = NotSet
     was_not: Any = NotSet
     changes_to: Any = NotSet
-    on_commit: bool = False
-    priority: int = DEFAULT_PRIORITY
+
+    def __post_init__(self):
+        super().__post_init__()
+
+        if self.condition is None:
+            self.condition = self._get_condition_from_legacy_parameters()
+
+    def _legacy_parameters_have_been_passed(self) -> bool:
+        return any(
+            [
+                self.when is not None,
+                self.when_any is not None,
+                self.was != "*",
+                self.is_now != "*",
+                self.has_changed is not None,
+                self.is_not is not NotSet,
+                self.was_not is not NotSet,
+                self.changes_to is not NotSet,
+            ]
+        )
+
+    def _get_condition_from_legacy_parameters(self) -> Callable:
+        if self.when:
+            return When(
+                when=self.when,
+                was=self.was,
+                is_now=self.is_now,
+                has_changed=self.has_changed,
+                is_not=self.is_not,
+                was_not=self.was_not,
+                changes_to=self.changes_to,
+            )
+
+        elif self.when_any:
+            return reduce(
+                operator.or_,
+                [
+                    When(
+                        when=field,
+                        was=self.was,
+                        is_now=self.is_now,
+                        has_changed=self.has_changed,
+                        is_not=self.is_not,
+                        was_not=self.was_not,
+                        changes_to=self.changes_to,
+                    )
+                    for field in self.when_any
+                ],
+            )
+        else:
+            return Always()
 
     def validate_hook(self, value, **kwargs):
         if value not in VALID_HOOKS:
@@ -101,9 +163,16 @@ def validate_when_and_when_any(self):
                 "Can pass either 'when' or 'when_any' but not both"
             )
 
+    def validate_condition_and_legacy_parameters_are_not_combined(self):
+        if self.condition is not None and self._legacy_parameters_have_been_passed():
+            raise DjangoLifeCycleException(
+                "Legacy parameters (when, when_any, ...) can't be used together with condition"
+            )
+
     def validate(self):
         self.validate_when_and_when_any()
         self.validate_on_commit_only_for_after_hooks()
+        self.validate_condition_and_legacy_parameters_are_not_combined()
 
     def __lt__(self, other):
         if not isinstance(other, HookConfig):