diff --git a/python/lib/sift_py/rule/config.py b/python/lib/sift_py/rule/config.py index 5729c4d0a..487f32369 100644 --- a/python/lib/sift_py/rule/config.py +++ b/python/lib/sift_py/rule/config.py @@ -12,6 +12,88 @@ from sift_py.ingestion.channel import ChannelConfig +class RuleAction(ABC): + tags: Optional[List[str]] + + @abstractmethod + def kind(self) -> RuleActionKind: + pass + + +class RuleActionCreateDataReviewAnnotation(RuleAction): + """ + Action to create a data-review annotation when a rule evaluates to a truthy value. + + - `tags`: List of tag names to associate with the newly created data-review annotation. + - `assignee`: Email of user in organization to assign the newly created data-review annotation. + """ + + tags: Optional[List[str]] + assignee: Optional[str] + + def __init__(self, assignee: Optional[str] = None, tags: Optional[List[str]] = None): + self.assignee = assignee + self.tags = tags + + def kind(self) -> RuleActionKind: + return RuleActionKind.ANNOTATION + + +class RuleActionCreatePhaseAnnotation(RuleAction): + """ + Action to create a phase annotation when a rule evaluates to a truthy value. + + - `tags`: List of tag names to associate with the newly created data-review annotation. + """ + + tags: Optional[List[str]] + + def __init__(self, tags: Optional[List[str]] = None): + self.tags = tags + + def kind(self) -> RuleActionKind: + return RuleActionKind.ANNOTATION + + +class RuleActionKind(Enum): + NOTIFICATION = ActionKind.NOTIFICATION + ANNOTATION = ActionKind.ANNOTATION + + @classmethod + def from_str(cls, val: str) -> Optional["RuleActionKind"]: + if val == "ACTION_KIND_NOTIFICATION" or val == RuleActionKindStrRep.NOTIFICATION.value: + return cls.NOTIFICATION + elif val == "ACTION_KIND_ANNOTATION" or val == RuleActionKindStrRep.ANNOTATION.value: + return cls.ANNOTATION + + return None + + +class RuleActionAnnotationKind(Enum): + REVIEW = "review" + PHASE = "phase" + + @classmethod + def from_annotation_type(cls, annotation_type: AnnotationType) -> "RuleActionAnnotationKind": + if annotation_type == AnnotationType.ANNOTATION_TYPE_PHASE: + return cls.PHASE + return cls.REVIEW + + @classmethod + def from_str(cls, val: str) -> "RuleActionAnnotationKind": + if val == cls.REVIEW.value: + return cls.REVIEW + elif val == cls.PHASE.value: + return cls.PHASE + else: + raise ValueError(f"Argument '{val}' is not a valid annotation kind.") + + +class RuleActionKindStrRep(Enum): + NOTIFICATION = "notification" + ANNOTATION = "annotation" + + class RuleConfig(AsJson): """ Defines a rule to be used during ingestion. If a rule's expression validates to try, then @@ -20,7 +102,7 @@ class RuleConfig(AsJson): - `name`: Name of the rule. - `description`: Description of the rule. - `expression`: A CEL string expression that executes the `action` when evaluated to a truthy value. - - `action`: The action to execute if the result of an `expression` evaluates to a truthy value. + - `action`: The action to execute if the result of an `expression` evaluates to a truthy value. Default is to create a data review annotation. - `channel_references`: Reference to channel. If an expression is "$1 < 10", then "$1" is the reference and thus should the key in the dict. - `rule_client_key`: User defined unique string that uniquely identifies this rule. - `asset_names`: A list of asset names that this rule should be applied to. ONLY VALID if defining rules outside of a telemetry config. @@ -34,7 +116,7 @@ class RuleConfig(AsJson): name: str description: str expression: str - action: Optional[RuleAction] + action: RuleAction channel_references: List[ExpressionChannelReference] rule_client_key: Optional[str] asset_names: List[str] @@ -52,7 +134,7 @@ def __init__( ], description: str = "", expression: str = "", - action: Optional[RuleAction] = None, + action: RuleAction = RuleActionCreateDataReviewAnnotation(), rule_client_key: Optional[str] = None, asset_names: Optional[List[str]] = None, tag_names: Optional[List[str]] = None, @@ -134,88 +216,6 @@ def interpolate_sub_expressions( return expression -class RuleAction(ABC): - tags: Optional[List[str]] - - @abstractmethod - def kind(self) -> RuleActionKind: - pass - - -class RuleActionCreateDataReviewAnnotation(RuleAction): - """ - Action to create a data-review annotation when a rule evaluates to a truthy value. - - - `tags`: List of tag names to associate with the newly created data-review annotation. - - `assignee`: Email of user in organization to assign the newly created data-review annotation. - """ - - tags: Optional[List[str]] - assignee: Optional[str] - - def __init__(self, assignee: Optional[str] = None, tags: Optional[List[str]] = None): - self.assignee = assignee - self.tags = tags - - def kind(self) -> RuleActionKind: - return RuleActionKind.ANNOTATION - - -class RuleActionCreatePhaseAnnotation(RuleAction): - """ - Action to create a phase annotation when a rule evaluates to a truthy value. - - - `tags`: List of tag names to associate with the newly created data-review annotation. - """ - - tags: Optional[List[str]] - - def __init__(self, tags: Optional[List[str]] = None): - self.tags = tags - - def kind(self) -> RuleActionKind: - return RuleActionKind.ANNOTATION - - -class RuleActionKind(Enum): - NOTIFICATION = ActionKind.NOTIFICATION - ANNOTATION = ActionKind.ANNOTATION - - @classmethod - def from_str(cls, val: str) -> Optional["RuleActionKind"]: - if val == "ACTION_KIND_NOTIFICATION" or val == RuleActionKindStrRep.NOTIFICATION.value: - return cls.NOTIFICATION - elif val == "ACTION_KIND_ANNOTATION" or val == RuleActionKindStrRep.ANNOTATION.value: - return cls.ANNOTATION - - return None - - -class RuleActionAnnotationKind(Enum): - REVIEW = "review" - PHASE = "phase" - - @classmethod - def from_annotation_type(cls, annotation_type: AnnotationType) -> "RuleActionAnnotationKind": - if annotation_type == AnnotationType.ANNOTATION_TYPE_PHASE: - return cls.PHASE - return cls.REVIEW - - @classmethod - def from_str(cls, val: str) -> "RuleActionAnnotationKind": - if val == cls.REVIEW.value: - return cls.REVIEW - elif val == cls.PHASE.value: - return cls.PHASE - else: - raise ValueError(f"Argument '{val}' is not a valid annotation kind.") - - -class RuleActionKindStrRep(Enum): - NOTIFICATION = "notification" - ANNOTATION = "annotation" - - class ExpressionChannelReference(TypedDict): """ `channel_reference`: The channel reference (e.g. '$1') used in the expression.