Use rule_evaluation protos for rule evaluation.

Author: ian-sift

python/lib/sift_client/_internal/low_level_wrappers/rules.py

@@ -3,6 +3,13 @@
 import logging
 from typing import TYPE_CHECKING, Any, cast
 
+from sift.common.type.v1.resource_identifier_pb2 import ResourceIdentifier
+from sift.rule_evaluation.v1.rule_evaluation_pb2 import (
+    EvaluateRulesRequest,
+    EvaluateRulesResponse,
+    RunTimeRange,
+)
+from sift.rule_evaluation.v1.rule_evaluation_pb2_grpc import RuleEvaluationServiceStub
 from sift.rules.v1.rules_pb2 import (
     BatchDeleteRulesRequest,
     BatchGetRulesRequest,
@@ -31,15 +38,20 @@
 from sift.rules.v1.rules_pb2_grpc import RuleServiceStub
 
 from sift_client._internal.low_level_wrappers.base import LowLevelClientBase
+from sift_client._internal.low_level_wrappers.reports import ReportsLowLevelClient
 from sift_client.sift_types.rule import (
     Rule,
     RuleAction,
     RuleUpdate,
 )
 from sift_client.transport import GrpcClient, WithGrpcClient
+from sift_client.util.util import count_non_none
 
 if TYPE_CHECKING:
+    from datetime import datetime
+
     from sift_client.sift_types.channel import ChannelReference
+    from sift_client.sift_types.report import Report
 
 # Configure logging
 logger = logging.getLogger(__name__)
@@ -445,3 +457,76 @@ async def list_all_rules(
             order_by=order_by,
             max_results=max_results,
         )
+
+    async def evaluate_rules(
+        self,
+        *,
+        run_id: str | None = None,
+        assets: list[str] | None = None,
+        all_applicable_rules: bool | None = None,
+        run_start_time: datetime | None = None,
+        run_end_time: datetime | None = None,
+        rule_ids: list[str] | None = None,
+        rule_version_ids: list[str] | None = None,
+        report_template_id: str | None = None,
+        tags: list[str] | None = None,
+    ) -> Report | None:
+        """Evaluate a rule.
+
+        Args:
+            run_id: The run ID to evaluate.
+            assets: The assets to evaluate.
+            run_start_time: The start time of the run.
+            run_end_time: The end time of the run.
+            all_applicable_rules: Whether to evaluate all rules applicable to the selected run, assets, or time range.
+            rule_ids: The rule IDs to evaluate.
+            rule_version_ids: The rule version IDs to evaluate.
+            report_template_id: The report template ID to evaluate.
+            tags: Optional tags to add to generated annotations.
+
+        Returns:
+            The result of the rule execution.
+        """
+        if count_non_none(run_id, assets, run_start_time, run_end_time) > 1:
+            raise ValueError(
+                "Pick only one run_id, assets, or (run_start_time and run_end_time) to select what to evaluate against."
+            )
+
+        all_applicable_rules = (
+            None if not all_applicable_rules else True
+        )  # Cast to None if False so we don't count it against other filters if they aren't opting in.
+        if count_non_none(rule_ids, rule_version_ids, report_template_id, all_applicable_rules) > 1:
+            raise ValueError(
+                "Pick only one rule_ids, rule_version_ids, report_template_id, or all_applicable_rules to further filter which rules to evaluate."
+            )
+
+        kwargs: dict[str, Any] = {}
+        if run_start_time and run_end_time:
+            kwargs["run_time_range"] = RunTimeRange(
+                run=run_id, start_time=run_start_time, end_time=run_end_time
+            )
+        if run_id:
+            kwargs["run"] = ResourceIdentifier(id=run_id)
+        if assets:
+            kwargs["assets"] = assets
+        if all_applicable_rules:
+            kwargs["all_applicable_rules"] = all_applicable_rules
+        if rule_ids:
+            kwargs["rules"] = rule_ids
+        if rule_version_ids:
+            kwargs["rule_versions"] = rule_version_ids
+        if report_template_id:
+            kwargs["report_template"] = report_template_id
+        if tags:
+            kwargs["tags"] = tags
+
+        request = EvaluateRulesRequest(**kwargs)
+        response = await self._grpc_client.get_stub(RuleEvaluationServiceStub).EvaluateRules(
+            request
+        )
+        response = cast("EvaluateRulesResponse", response)
+        report_id = response.report_id
+        if report_id:
+            report = await ReportsLowLevelClient(self._grpc_client).get_report(report_id=report_id)
+            return report
+        return None

python/lib/sift_client/_internal/low_level_wrappers/tags.py

@@ -12,7 +12,7 @@
 from sift.tags.v2.tags_pb2_grpc import TagServiceStub
 
 from sift_client._internal.low_level_wrappers.base import LowLevelClientBase
-from sift_client.sift_types.tag import Tag, TagUpdate
+from sift_client.sift_types.tag import Tag
 from sift_client.transport import WithGrpcClient
 
 if TYPE_CHECKING:
@@ -114,4 +114,4 @@ async def list_all_tags(
             kwargs={"query_filter": query_filter},
             order_by=order_by,
             max_results=max_results,
-        )
+        )

python/lib/sift_client/_tests/integrated/reports.py

@@ -9,11 +9,11 @@
 
 import asyncio
 import os
-from datetime import datetime, timedelta, timezone
+from datetime import datetime
+
 from zoneinfo import ZoneInfo
 
 from sift_client import SiftClient
-from sift_client.sift_types import report
 
 
 async def main():
@@ -23,6 +23,7 @@ async def main():
     grpc_url = os.getenv("SIFT_GRPC_URI", "localhost:50051")
     rest_url = os.getenv("SIFT_REST_URI", "localhost:8080")
     api_key = os.getenv("SIFT_API_KEY", "")
+    
     client = SiftClient(
         api_key=api_key,
         grpc_url=grpc_url,
@@ -35,49 +36,24 @@ async def main():
         limit=100,
     )
 
-    asset_ids = []
-    asset_tags_names = []
     rules = []
-    reports = []
+    failed_runs = []
     for run in runs:
         print("run.name: ", run.name)
         print(" client_key: ", run.client_key)
-        if run.client_key:
-            # rules = client.rules.list_(
-            #     client_key=run.client_key,
-            #     limit=100,
-            # )
-            raise Exception("client_key is not None! Let's add these rules")
-        run_assets = run.assets
-        print("  assets: ", [asset.name for asset in run_assets])
-        asset_ids.extend([asset.id_ for asset in run_assets])
-        asset_tags_names.extend([tag for asset in run_assets for tag in asset.tags])
-        per_run_reports = client.reports.list_(
-            run_id=run.id_,
-        )
-        print("  reports: ", [report.name for report in per_run_reports])
-        reports.extend(per_run_reports)  
-    
-    asset_ids = list(set(asset_ids))
-    asset_tags_names = list(set(asset_tags_names))
-    asset_tags = client.tags.list_(
-        names=asset_tags_names,
-    )
-    print("  asset_tags: ", [(tag.name, tag.id_) for tag in asset_tags])
-    print("Number of runs: ", len(runs))
-    print("Number of assets: ", len(asset_ids))
+        try:
+            report = client.rules.evaluate(
+                run_id=run.id_,
+                all_applicable_rules=True,
+            )
+        except Exception as e:
+            failed_runs.append(run.id_)
+            print(f"Failed to evaluate rules for run {run.id_}: {e}")
+
+    print("Number of successful runs: ", len(runs) - len(failed_runs))
+    print("Number of failed runs: ", len(failed_runs))
 
-    rules = client.rules.list_(
-        asset_ids=asset_ids,
-        asset_tags_ids=[tag.id_ for tag in asset_tags],
-    )
-    print("reports: ", [report.name for report in reports])
-    if len(rules) < 10:
-        print("rules: ", [rule.name for rule in rules])
-    else:
-        print("number of rules: ", len(rules))
 
-    
 
 if __name__ == "__main__":
     asyncio.run(main())

python/lib/sift_client/resources/__init__.py

@@ -7,6 +7,8 @@
 from sift_client.resources.rules import RulesAPIAsync
 from sift_client.resources.runs import RunsAPIAsync
 from sift_client.resources.tags import TagsAPIAsync
+
+# ruff: noqa TagsAPIAsync needs to be imported before sync_stubs to avoid circular import
 from sift_client.resources.sync_stubs import (
     AssetsAPI,
     CalculatedChannelsAPI,

python/lib/sift_client/resources/reports.py

@@ -6,15 +6,14 @@
 from sift_client._internal.low_level_wrappers.reports import ReportsLowLevelClient
 from sift_client.resources._base import ResourceBase
 from sift_client.sift_types.report import Report
-from sift_client.util.cel_utils import contains, equals, equals_null, match, not_
+from sift_client.util.cel_utils import contains, equals, match
 
 if TYPE_CHECKING:
     from sift_client.client import SiftClient
 
 
 class ReportsAPIAsync(ResourceBase):
-    """High-level API for interacting with reports.
-    """
+    """High-level API for interacting with reports."""
 
     def __init__(self, sift_client: SiftClient):
         """Initialize the ReportsAPI.
@@ -201,7 +200,6 @@ async def create_from_rules(
         )
         return self._apply_client_to_instance(created_report)
 
-
     async def rerun(
         self,
         *,
@@ -233,4 +231,4 @@ async def cancel(
         report_id = report.id_ if isinstance(report, Report) else report
         if not isinstance(report_id, str):
             raise TypeError(f"report_id must be a string not {type(report_id)}")
-        await self._low_level_client.cancel_report(report_id=report_id)
+        await self._low_level_client.cancel_report(report_id=report_id)

python/lib/sift_client/resources/rules.py

@@ -9,9 +9,11 @@
 
 if TYPE_CHECKING:
     import re
+    from datetime import datetime
 
     from sift_client.client import SiftClient
     from sift_client.sift_types.channel import ChannelReference
+    from sift_client.sift_types.report import Report
 
 
 class RulesAPIAsync(ResourceBase):
@@ -273,3 +275,56 @@ async def batch_get(
             rule_ids=rule_ids, client_keys=client_keys
         )
         return self._apply_client_to_instances(rules)
+
+    async def evaluate(
+        self,
+        *,
+        run_id: str | None = None,
+        assets: list[str] | None = None,
+        all_applicable_rules: bool | None = None,
+        run_start_time: datetime | None = None,
+        run_end_time: datetime | None = None,
+        rule_ids: list[str] | None = None,
+        rule_version_ids: list[str] | None = None,
+        report_template_id: str | None = None,
+        tags: list[str] | None = None,
+    ) -> Report | None:
+        """Evaluate a rule.
+
+        Pick one of the following grouping of rules to evaluate against:
+        - run_id
+        - assets
+        - run_start_time and run_end_time
+        And one of the following filters to select which rules to evaluate:
+        - rule_ids
+        - rule_version_ids
+        - report_template_id
+        - all_applicable_rules
+
+        Args:
+            run_id: The run ID to evaluate.
+            assets: The assets to evaluate.
+            all_applicable_rules: Whether to evaluate all rules applicable to the selected run, assets, or time range.
+            run_start_time: The start time of the run.
+            run_end_time: The end time of the run.
+            rule_ids: The rule IDs to evaluate.
+            rule_version_ids: The rule version IDs to evaluate.
+            report_template_id: The report template ID to evaluate.
+            tags: Optional tags to add to generated annotations.
+
+        Returns:
+            The result of the rule evaluation.
+        """
+        report = await self._low_level_client.evaluate_rules(
+            run_id=run_id,
+            assets=assets,
+            all_applicable_rules=all_applicable_rules,
+            run_start_time=run_start_time,
+            run_end_time=run_end_time,
+            rule_ids=rule_ids,
+            rule_version_ids=rule_version_ids,
+            report_template_id=report_template_id,
+            tags=tags,
+        )
+        if report:
+            return self._apply_client_to_instance(report)

python/lib/sift_client/resources/runs.py

@@ -6,7 +6,16 @@
 from sift_client._internal.low_level_wrappers.runs import RunsLowLevelClient
 from sift_client.resources._base import ResourceBase
 from sift_client.sift_types.run import Run, RunUpdate
-from sift_client.util.cel_utils import contains, equals, equals_null, greater_than, less_than, match, not_
+from sift_client.util.cel_utils import (
+    contains,
+    equals,
+    equals_null,
+    greater_than,
+    in_,
+    less_than,
+    match,
+    not_,
+)
 
 if TYPE_CHECKING:
     from datetime import datetime
@@ -55,6 +64,7 @@ async def list_(
         name: str | None = None,
         name_contains: str | None = None,
         name_regex: str | re.Pattern | None = None,
+        run_ids: list[str] | None = None,
         description: str | None = None,
         description_contains: str | None = None,
         duration_seconds: int | None = None,
@@ -82,6 +92,7 @@ async def list_(
             name: Exact name of the run.
             name_contains: Partial name of the run.
             name_regex: Regular expression string to filter runs by name.
+            run_ids: List of run IDs to filter by.
             description: Exact description of the run.
             description_contains: Partial description of the run.
             duration_seconds: Duration of the run in seconds.
@@ -118,6 +129,9 @@ async def list_(
                 name_regex = name_regex.pattern
             filter_parts.append(match("name", name_regex))  # type: ignore
 
+        if run_ids:
+            filter_parts.append(in_("run_id", run_ids))
+
         if description:
             filter_parts.append(equals("description", description))
         elif description_contains: