sift_py.rule.config

CLASS	DESCRIPTION
ExpressionChannelReference	channel_reference: The channel reference (e.g. '$1') used in the expression.
ExpressionChannelReferenceChannelConfig	channel_reference: The channel reference (e.g. '$1') used in the expression.
RuleAction
RuleActionAnnotationKind
RuleActionCreateDataReviewAnnotation	Action to create a data-review annotation when a rule evaluates to a truthy value.
RuleActionCreatePhaseAnnotation	Action to create a phase annotation when a rule evaluates to a truthy value.
RuleActionKind
RuleActionKindStrRep
RuleConfig	Defines a rule to be used during ingestion. If a rule's expression validates to try, then

ExpressionChannelReference

Bases: TypedDict

channel_reference: The channel reference (e.g. '$1') used in the expression. channel_identifier: The channel name.

ATTRIBUTE	DESCRIPTION
channel_identifier	TYPE: str
channel_reference	TYPE: str

channel_identifier instance-attribute

channel_identifier: str

channel_reference instance-attribute

channel_reference: str

ExpressionChannelReferenceChannelConfig

Bases: TypedDict

channel_reference: The channel reference (e.g. '$1') used in the expression. channel_config: Instance of sift_py.ingestion.channel.ChannelConfig.

ATTRIBUTE	DESCRIPTION
channel_config	TYPE: ChannelConfig
channel_reference	TYPE: str

channel_config instance-attribute

channel_config: ChannelConfig

channel_reference instance-attribute

channel_reference: str

RuleAction

Bases: ABC

METHOD	DESCRIPTION
kind

kind abstractmethod

kind() -> RuleActionKind

RuleActionAnnotationKind

Bases: Enum

METHOD	DESCRIPTION
from_annotation_type
from_str

ATTRIBUTE	DESCRIPTION
PHASE
REVIEW

PHASE class-attribute instance-attribute

PHASE = 'phase'

REVIEW class-attribute instance-attribute

REVIEW = 'review'

from_annotation_type classmethod

from_annotation_type(
    annotation_type: AnnotationType,
) -> "RuleActionAnnotationKind"

from_str classmethod

from_str(val: str) -> 'RuleActionAnnotationKind'

RuleActionCreateDataReviewAnnotation

RuleActionCreateDataReviewAnnotation(
    assignee: Optional[str] = None,
    tags: Optional[List[str]] = None,
)

Bases: 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.

METHOD	DESCRIPTION
kind

ATTRIBUTE	DESCRIPTION
assignee	TYPE: Optional[str]
tags	TYPE: Optional[List[str]]

assignee instance-attribute

assignee: Optional[str] = assignee

tags instance-attribute

tags: Optional[List[str]] = tags

kind

kind() -> RuleActionKind

RuleActionCreatePhaseAnnotation

RuleActionCreatePhaseAnnotation(
    tags: Optional[List[str]] = None,
)

Bases: 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.

METHOD	DESCRIPTION
kind

ATTRIBUTE	DESCRIPTION
tags	TYPE: Optional[List[str]]

tags instance-attribute

tags: Optional[List[str]] = tags

kind

kind() -> RuleActionKind

RuleActionKind

Bases: Enum

METHOD	DESCRIPTION
from_str

ATTRIBUTE	DESCRIPTION
ANNOTATION
NOTIFICATION

ANNOTATION class-attribute instance-attribute

ANNOTATION = ANNOTATION

NOTIFICATION class-attribute instance-attribute

NOTIFICATION = NOTIFICATION

from_str classmethod

from_str(val: str) -> Optional['RuleActionKind']

RuleActionKindStrRep

Bases: Enum

ATTRIBUTE	DESCRIPTION
ANNOTATION
NOTIFICATION

ANNOTATION class-attribute instance-attribute

ANNOTATION = 'annotation'

NOTIFICATION class-attribute instance-attribute

NOTIFICATION = 'notification'

RuleConfig

RuleConfig(
    name: str,
    channel_references: List[
        Union[
            ExpressionChannelReference,
            ExpressionChannelReferenceChannelConfig,
        ]
    ],
    description: str = "",
    expression: str = "",
    action: Optional[RuleAction] = None,
    rule_client_key: Optional[str] = None,
    asset_names: Optional[List[str]] = None,
    tag_names: Optional[List[str]] = None,
    sub_expressions: Dict[str, Any] = {},
    contextual_channels: Optional[List[str]] = None,
    is_external: bool = False,
)

Bases: AsJson

Defines a rule to be used during ingestion. If a rule's expression validates to try, then a specific action will take place as specified by the kind attribute.

• 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.
• 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.
• tag_names: A list of asset names that this rule should be applied to. ONLY VALID if defining rules outside of a telemetry config.
• contextual_channels: A list of channel names that provide context but aren't directly used in the expression.
• is_external: If this is an external rule.

METHOD	DESCRIPTION
as_json	Produces the appropriate JSON structure that's suitable for the Rules API.
interpolate_sub_expressions

ATTRIBUTE	DESCRIPTION
action	TYPE: Optional[RuleAction]
asset_names	TYPE: List[str]
channel_references	TYPE: List[ExpressionChannelReference]
contextual_channels	TYPE: List[str]
description	TYPE: str
expression	TYPE: str
is_external	TYPE: bool
name	TYPE: str
rule_client_key	TYPE: Optional[str]

action instance-attribute

action: Optional[RuleAction] = action

asset_names instance-attribute

asset_names: List[str] = asset_names or []

channel_references instance-attribute

channel_references: List[ExpressionChannelReference] = (
    _channel_references_from_dicts(channel_references)
)

contextual_channels instance-attribute

contextual_channels: List[str] = contextual_channels or []

description instance-attribute

description: str = description

expression instance-attribute

expression: str = interpolate_sub_expressions(
    expression, sub_expressions
)

is_external instance-attribute

is_external: bool = is_external

name instance-attribute

name: str = name

rule_client_key instance-attribute

rule_client_key: Optional[str] = rule_client_key

as_json

as_json() -> Any

Produces the appropriate JSON structure that's suitable for the Rules API.

interpolate_sub_expressions staticmethod

interpolate_sub_expressions(
    expression: str,
    sub_expressions: Optional[Dict[str, str]],
) -> str