sift_py.ingestion.rule.config

 16class RuleConfig(AsJson):
 17    """
 18    Defines a rule to be used during ingestion. If a rule's expression validates to try, then
 19    a specific action will take place as specified by the `kind` attribute.
 20
 21    - `name`: Name of the rule.
 22    - `description`: Description of the rule.
 23    - `expression`: A CEL string expression that executes the `action` when evaluated to a truthy value.
 24    - `action`: The action to execute if the result of an `expression` evaluates to a truthy value.
 25    - `channel_references`: Reference to channel. If an expression is "$1 < 10", then "$1" is the reference and thus should the key in the dict.
 26    - `rule_client_key`: User defined unique string that uniquely identifies this rule.
 27    - `asset_names`: A list of asset names that this rule should be applied to. ONLY VALID if defining rules outside of a telemetry config.
 28    - `tag_names`: A list of asset names that this rule should be applied to. ONLY VALID if defining rules outside of a telemetry config.
 29    - `contextual_channels`: A list of channel names that provide context but aren't directly used in the expression.
 30    - `is_external`: If this is an external rule.
 31    """
 32
 33    name: str
 34    description: str
 35    expression: str
 36    action: Optional[RuleAction]
 37    channel_references: List[ExpressionChannelReference]
 38    rule_client_key: Optional[str]
 39    asset_names: List[str]
 40    contextual_channels: List[str]
 41    is_external: bool
 42
 43    def __init__(
 44        self,
 45        name: str,
 46        channel_references: List[
 47            Union[ExpressionChannelReference, ExpressionChannelReferenceChannelConfig]
 48        ],
 49        description: str = "",
 50        expression: str = "",
 51        action: Optional[RuleAction] = None,
 52        rule_client_key: Optional[str] = None,
 53        asset_names: Optional[List[str]] = None,
 54        tag_names: Optional[List[str]] = None,
 55        sub_expressions: Dict[str, Any] = {},
 56        contextual_channels: Optional[List[str]] = None,
 57        is_external: bool = False,
 58    ):
 59        self.channel_references = _channel_references_from_dicts(channel_references)
 60        self.contextual_channels = contextual_channels or []
 61
 62        self.name = name
 63        self.asset_names = asset_names or []
 64        self.action = action
 65        self.rule_client_key = rule_client_key
 66        self.description = description
 67        self.expression = self.__class__.interpolate_sub_expressions(expression, sub_expressions)
 68        self.is_external = is_external
 69
 70    def as_json(self) -> Any:
 71        """
 72        Produces the appropriate JSON structure that's suitable for the Rules API.
 73        """
 74
 75        hash_map: Dict[
 76            str,
 77            Union[
 78                List[ExpressionChannelReference], List[ChannelConfig], str, List[str], bool, None
 79            ],
 80        ] = {
 81            "name": self.name,
 82            "description": self.description,
 83            "expression": self.expression,
 84            "is_external": self.is_external,
 85        }
 86
 87        hash_map["expression_channel_references"] = self.channel_references
 88        if self.contextual_channels:
 89            hash_map["contextual_channel_references"] = self.contextual_channels
 90
 91        if isinstance(self.action, RuleActionCreateDataReviewAnnotation):
 92            hash_map["type"] = RuleActionAnnotationKind.REVIEW.value
 93            hash_map["assignee"] = self.action.assignee
 94
 95            if self.action.assignee is not None and len(self.action.assignee) > 0:
 96                hash_map["assignee"] = self.action.assignee
 97
 98            if self.action.tags is not None and len(self.action.tags) > 0:
 99                hash_map["tags"] = self.action.tags
100
101        elif isinstance(self.action, RuleActionCreatePhaseAnnotation):
102            hash_map["type"] = RuleActionAnnotationKind.PHASE.value
103
104            if self.action.tags is not None and len(self.action.tags) > 0:
105                hash_map["tags"] = self.action.tags
106        else:
107            kind = self.action.kind() if self.action else self.action
108            raise TypeError(f"Unsupported rule action '{kind}'.")
109
110        return hash_map
111
112    @staticmethod
113    def interpolate_sub_expressions(
114        expression: str, sub_expressions: Optional[Dict[str, str]]
115    ) -> str:
116        if sub_expressions:
117            for ref, expr in sub_expressions.items():
118                if ref not in expression:
119                    raise ValueError(f"Couldn't find '{ref}' in expression '{expression}'.")
120                if isinstance(expr, str):
121                    expression = expression.replace(ref, f'"{expr}"')
122                else:
123                    expression = expression.replace(ref, str(expr))
124
125        return expression

128class RuleAction(ABC):
129    @abstractmethod
130    def kind(self) -> RuleActionKind:
131        pass

134class RuleActionCreateDataReviewAnnotation(RuleAction):
135    """
136    Action to create a data-review annotation when a rule evaluates to a truthy value.
137
138    - `tags`: List of tag names to associate with the newly created data-review annotation.
139    - `assignee`: Email of user in organization to assign the newly created data-review annotation.
140    """
141
142    tags: Optional[List[str]]
143    assignee: Optional[str]
144
145    def __init__(self, assignee: Optional[str] = None, tags: Optional[List[str]] = None):
146        self.assignee = assignee
147        self.tags = tags
148
149    def kind(self) -> RuleActionKind:
150        return RuleActionKind.ANNOTATION

153class RuleActionCreatePhaseAnnotation(RuleAction):
154    """
155    Action to create a phase annotation when a rule evaluates to a truthy value.
156
157    - `tags`: List of tag names to associate with the newly created data-review annotation.
158    """
159
160    tags: Optional[List[str]]
161
162    def __init__(self, tags: Optional[List[str]] = None):
163        self.tags = tags
164
165    def kind(self) -> RuleActionKind:
166        return RuleActionKind.ANNOTATION

183class RuleActionAnnotationKind(Enum):
184    REVIEW = "review"
185    PHASE = "phase"
186
187    @classmethod
188    def from_annotation_type(cls, annotation_type: AnnotationType) -> "RuleActionAnnotationKind":
189        if annotation_type == AnnotationType.ANNOTATION_TYPE_PHASE:
190            return cls.PHASE
191        return cls.REVIEW
192
193    @classmethod
194    def from_str(cls, val: str) -> "RuleActionAnnotationKind":
195        if val == cls.REVIEW.value:
196            return cls.REVIEW
197        elif val == cls.PHASE.value:
198            return cls.PHASE
199        else:
200            raise ValueError(f"Argument '{val}' is not a valid annotation kind.")

169class RuleActionKind(Enum):
170    NOTIFICATION = ActionKind.NOTIFICATION
171    ANNOTATION = ActionKind.ANNOTATION
172
173    @classmethod
174    def from_str(cls, val: str) -> Optional["RuleActionKind"]:
175        if val == "ACTION_KIND_NOTIFICATION" or val == RuleActionKindStrRep.NOTIFICATION.value:
176            return cls.NOTIFICATION
177        elif val == "ACTION_KIND_ANNOTATION" or val == RuleActionKindStrRep.ANNOTATION.value:
178            return cls.ANNOTATION
179
180        return None

203class RuleActionKindStrRep(Enum):
204    NOTIFICATION = "notification"
205    ANNOTATION = "annotation"

208class ExpressionChannelReference(TypedDict):
209    """
210    `channel_reference`: The channel reference (e.g. '$1') used in the expression.
211    `channel_identifier`: The channel name.
212    """
213
214    channel_reference: str
215    channel_identifier: str

218class ExpressionChannelReferenceChannelConfig(TypedDict):
219    """
220    `channel_reference`: The channel reference (e.g. '$1') used in the expression.
221    `channel_config`: Instance of `sift_py.ingestion.channel.ChannelConfig`.
222    """
223
224    channel_reference: str
225    channel_config: ChannelConfig