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    """
 30
 31    name: str
 32    description: str
 33    expression: str
 34    action: Optional[RuleAction]
 35    channel_references: List[ExpressionChannelReference]
 36    rule_client_key: Optional[str]
 37    asset_names: List[str]
 38
 39    def __init__(
 40        self,
 41        name: str,
 42        channel_references: List[
 43            Union[ExpressionChannelReference, ExpressionChannelReferenceChannelConfig]
 44        ],
 45        description: str = "",
 46        expression: str = "",
 47        action: Optional[RuleAction] = None,
 48        rule_client_key: Optional[str] = None,
 49        asset_names: Optional[List[str]] = None,
 50        tag_names: Optional[List[str]] = None,
 51        sub_expressions: Dict[str, Any] = {},
 52    ):
 53        self.channel_references = _channel_references_from_dicts(channel_references)
 54
 55        self.name = name
 56        self.asset_names = asset_names or []
 57        self.action = action
 58        self.rule_client_key = rule_client_key
 59        self.description = description
 60        self.expression = self.__class__.interpolate_sub_expressions(expression, sub_expressions)
 61
 62    def as_json(self) -> Any:
 63        """
 64        Produces the appropriate JSON structure that's suitable for the Rules API.
 65        """
 66
 67        hash_map: Dict[str, Union[List[ExpressionChannelReference], str, List[str], None]] = {
 68            "name": self.name,
 69            "description": self.description,
 70            "expression": self.expression,
 71        }
 72
 73        hash_map["expression_channel_references"] = self.channel_references
 74
 75        if isinstance(self.action, RuleActionCreateDataReviewAnnotation):
 76            hash_map["type"] = RuleActionAnnotationKind.REVIEW.value
 77            hash_map["assignee"] = self.action.assignee
 78
 79            if self.action.assignee is not None and len(self.action.assignee) > 0:
 80                hash_map["assignee"] = self.action.assignee
 81
 82            if self.action.tags is not None and len(self.action.tags) > 0:
 83                hash_map["tags"] = self.action.tags
 84
 85        elif isinstance(self.action, RuleActionCreatePhaseAnnotation):
 86            hash_map["type"] = RuleActionAnnotationKind.PHASE.value
 87
 88            if self.action.tags is not None and len(self.action.tags) > 0:
 89                hash_map["tags"] = self.action.tags
 90        else:
 91            kind = self.action.kind() if self.action else self.action
 92            raise TypeError(f"Unsupported rule action '{kind}'.")
 93
 94        return hash_map
 95
 96    @staticmethod
 97    def interpolate_sub_expressions(
 98        expression: str, sub_expressions: Optional[Dict[str, str]]
 99    ) -> str:
100        if sub_expressions:
101            for ref, expr in sub_expressions.items():
102                if ref not in expression:
103                    raise ValueError(f"Couldn't find '{ref}' in expression '{expression}'.")
104                if isinstance(expr, str):
105                    expression = expression.replace(ref, f'"{expr}"')
106                else:
107                    expression = expression.replace(ref, str(expr))
108
109        return expression

112class RuleAction(ABC):
113    @abstractmethod
114    def kind(self) -> RuleActionKind:
115        pass

118class RuleActionCreateDataReviewAnnotation(RuleAction):
119    """
120    Action to create a data-review annotation when a rule evaluates to a truthy value.
121
122    - `tags`: List of tag names to associate with the newly created data-review annotation.
123    - `assignee`: Email of user in organization to assign the newly created data-review annotation.
124    """
125
126    tags: Optional[List[str]]
127    assignee: Optional[str]
128
129    def __init__(self, assignee: Optional[str] = None, tags: Optional[List[str]] = None):
130        self.assignee = assignee
131        self.tags = tags
132
133    def kind(self) -> RuleActionKind:
134        return RuleActionKind.ANNOTATION

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

167class RuleActionAnnotationKind(Enum):
168    REVIEW = "review"
169    PHASE = "phase"
170
171    @classmethod
172    def from_annotation_type(cls, annotation_type: AnnotationType) -> "RuleActionAnnotationKind":
173        if annotation_type == AnnotationType.ANNOTATION_TYPE_PHASE:
174            return cls.PHASE
175        return cls.REVIEW
176
177    @classmethod
178    def from_str(cls, val: str) -> "RuleActionAnnotationKind":
179        if val == cls.REVIEW.value:
180            return cls.REVIEW
181        elif val == cls.PHASE.value:
182            return cls.PHASE
183        else:
184            raise ValueError(f"Argument '{val}' is not a valid annotation kind.")

153class RuleActionKind(Enum):
154    NOTIFICATION = ActionKind.NOTIFICATION
155    ANNOTATION = ActionKind.ANNOTATION
156
157    @classmethod
158    def from_str(cls, val: str) -> Optional["RuleActionKind"]:
159        if val == "ACTION_KIND_NOTIFICATION" or val == RuleActionKindStrRep.NOTIFICATION.value:
160            return cls.NOTIFICATION
161        elif val == "ACTION_KIND_ANNOTATION" or val == RuleActionKindStrRep.ANNOTATION.value:
162            return cls.ANNOTATION
163
164        return None

187class RuleActionKindStrRep(Enum):
188    NOTIFICATION = "notification"
189    ANNOTATION = "annotation"

192class ExpressionChannelReference(TypedDict):
193    """
194    `channel_reference`: The channel reference (e.g. '$1') used in the expression.
195    `channel_identifier`: The channel name.
196    """
197
198    channel_reference: str
199    channel_identifier: str

202class ExpressionChannelReferenceChannelConfig(TypedDict):
203    """
204    `channel_reference`: The channel reference (e.g. '$1') used in the expression.
205    `channel_config`: Instance of `sift_py.ingestion.channel.ChannelConfig`.
206    """
207
208    channel_reference: str
209    channel_config: ChannelConfig