Reference: Reduce API calls when creating rules

python/lib/sift_py/_internal/channel.py

@@ -1,3 +1,4 @@
+import functools
 import warnings
 from typing import List, Optional, cast
 
@@ -15,6 +16,7 @@ def channel_fqn(name: str, component: Optional[str] = None) -> str:
     return name if component is None or len(component) == 0 else f"{component}.{name}"
 
 
+@functools.cache
 def get_channels(
     channel_service: ChannelServiceStub,
     filter: str,

python/lib/sift_py/_internal/user.py

@@ -1,10 +1,12 @@
+import functools
 from typing import List, cast
 
 from sift.common.type.v1.user_pb2 import User
 from sift.users.v2.users_pb2 import ListActiveUsersRequest, ListActiveUsersResponse
 from sift.users.v2.users_pb2_grpc import UserServiceStub
 
 
+@functools.cache
 def get_active_users(
     user_service: UserServiceStub,
     filter: str,

python/lib/sift_py/rule/service.py

@@ -1,8 +1,9 @@
 from __future__ import annotations
 
+import functools
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Union, cast
+from typing import Any, Dict, List, Optional, Tuple, Union, cast
 
 from sift.annotations.v1.annotations_pb2 import AnnotationType
 from sift.assets.v1.assets_pb2 import Asset
@@ -262,7 +263,7 @@ def detach_asset(self, rule: Union[str, RuleConfig], asset_names: List[str]) ->
     def _attach_or_detach_asset(
         self, rule: Union[str, RuleConfig], asset_names: List[str], attach: bool
     ) -> RuleConfig:
-        assets = self._get_assets(names=asset_names)
+        assets = self._get_assets(names=tuple(sorted(asset_names)))
         if not assets:
             raise ValueError(
                 f"Cannot find all assets in list '{asset_names}'. One of these assets does not exist."
@@ -388,7 +389,11 @@ def _update_req_from_rule_config(
             )
 
         # TODO: once we have TagService_ListTags we can do asset-agnostic rules via tags
-        assets = self._get_assets(names=config.asset_names) if config.asset_names else None
+        assets = (
+            self._get_assets(names=tuple(sorted(config.asset_names)))
+            if config.asset_names
+            else None
+        )
 
         actions = []
         if config.action.kind() == RuleActionKind.NOTIFICATION:
@@ -557,7 +562,7 @@ def get_rule(self, rule: str) -> Optional[RuleConfig]:
                     )
 
         assets = self._get_assets(
-            ids=[asset_id for asset_id in rule_pb.asset_configuration.asset_ids]
+            ids=tuple(sorted([asset_id for asset_id in rule_pb.asset_configuration.asset_ids]))
         )
         asset_names = [asset.name for asset in assets]
 
@@ -597,7 +602,8 @@ def _get_rule_from_rule_id(self, rule_id: str) -> Optional[Rule]:
         except:
             return None
 
-    def _get_assets(self, names: List[str] = [], ids: List[str] = []) -> List[Asset]:
+    @functools.cache
+    def _get_assets(self, names: Tuple[str] = [], ids: Tuple[str] = []) -> List[Asset]:
         return list_assets_impl(self._asset_service_stub, names, ids)