update to filter_rule

.github/workflows/generate_scorecards_reminders.yaml

@@ -21,4 +21,4 @@ jobs:
                 slack_webhook_url: ${{ secrets.SLACK_WEBHOOK_URL }}
                 blueprint: ${{ secrets.BLUEPRINT }}
                 scorecard: ${{ secrets.SCORECARD }}
-                team: ${{ secrets.TEAM }}
+                filter_rule: '{"property": "$team","operator": "containsAny","value": ["Backend"]}'

README.md

@@ -22,5 +22,5 @@ See [action.yml](action.yml) for inputs and outputs.
     slack_webhook_url: ${{ secrets.SLACK_WEBHOOK_URL }}
     blueprint: service
     scorecard: Ownership
-    team: BACKEND-TEAM
+    filter_rule: '{"property": "$team","operator": "containsAny","value": ["Backend"]}'
 ```

action.yml

@@ -6,23 +6,23 @@ branding:
   color: gray-dark
 inputs:
   port_client_id:
-    description: "TThe Port Client ID to use to authenticate with the API"
+    description: 'The Port Client ID to use to authenticate with the API'
     required: true
   port_client_secret:
-    description: "The Port Client Secret to use to authenticate with the API"
+    description: 'The Port Client Secret to use to authenticate with the API'
     required: true
   slack_webhook_url:
-    description: "The URL of the Slack channel webhook to use to send messages"
+    description: 'The URL of the Slack channel webhook to use to send messages'
     required: true
   blueprint:
-    description: "The identifier of the blueprint in Port which you want to send reminders and reports about"
+    description: 'The identifier of the blueprint in Port which you want to send reminders and reports about'
     required: true
   scorecard:
-    description: "The scorecard to use to send reminders and reports about"
-    required: true
-  team:
-    description: "The team to use to filter by in Port, to get only the team resources you want to send reminders and reports about"
+    description: 'The scorecard to use to send reminders and reports about'
     required: true
+  filter_rule:
+    description: 'filter to use to filter the response, for example: {"property": "$team","operator": "containsAny","value": ["Backend"]}'
+    required: false
 runs:
   using: docker
   image: Dockerfile

config.py

@@ -1,5 +1,7 @@
 from enum import Enum
+from typing import List, Union, Optional
 
+from pydantic import Field, BaseModel
 from pydantic_settings import BaseSettings
 
 
@@ -11,14 +13,20 @@ class TargetKind(str, Enum):
     slack = "slack"
 
 
+class FilterRule(BaseModel):
+    property: str
+    operator: str
+    value: Union[str, int, List[str], List[int]] = None
+
+
 class Settings(BaseSettings):
     port_api_url: str = "https://api.getport.io"
     port_client_id: str
     port_client_secret: str
     slack_webhook_url: str
-    team: str
     blueprint: str
     scorecard: str
+    filter_rule: Optional[FilterRule] = Field(default=None)
     message_kind: MessageKind = MessageKind.generate_scorecards_reminders
     target_kind: TargetKind = TargetKind.slack
 

core/handler.py

@@ -15,31 +15,31 @@ class Handler:
     def generate_scorecards_reminders(cls):
         logger.info("Initializing Port client")
         port_client = PortClient(settings.port_api_url, settings.port_client_id, settings.port_client_secret)
-        logger.info(f"Fetching entities for team {settings.team}, blueprint {settings.blueprint}, scorecard {settings.scorecard}")
-        entities = port_client.search_entities(
-            {
+        logger.info(f"Fetching entities for query: {settings.filter_rule}, blueprint {settings.blueprint}, scorecard {settings.scorecard}")
+        search_query = {
                 "combinator": "and",
                 "rules": [
                     {
                         "property": "$blueprint",
                         "operator": "=",
                         "value": settings.blueprint
-                    },
-                    {
-                        "property": "$team",
-                        "operator": "containsAny",
-                        "value": [settings.team]
                     }
-                ]
-            }
+                ],
+        }
+        if settings.filter_rule:
+            search_query["rules"].append(settings.filter_rule.dict())
+
+        entities = port_client.search_entities(
+            search_query
         )
+        scorecard = port_client.get_scorecard(settings.blueprint, settings.scorecard).get("scorecard")
         if not entities:
             logger.info("No entities found")
             return
         if settings.target_kind == "slack":
             logger.info(f"Generating scorecards reminders for {len(entities)} entities")
             blocks = SlackMessageGenerator().generate_scorecards_reminders(settings.blueprint,
-                                                                           settings.scorecard,
+                                                                           scorecard,
                                                                            entities)
             logger.info("Sending scorecards reminders to slack channel")
             Slack().send_message(blocks)

generators/slack.py

@@ -8,7 +8,7 @@ class SlackMessageGenerator(generators.base.BaseMessageGenerator):
 
     def generate_scorecards_reminders(self,
                                       blueprint: str,
-                                      scorecard_name: str,
+                                      scorecard: Dict[str, Any],
                                       entities: list) -> List[Dict[str, Any]]:
         blueprint_plural = utils.convert_to_plural(blueprint)
         entities_didnt_pass_gold_level = {
@@ -18,11 +18,11 @@ def generate_scorecards_reminders(self,
         }
         number_of_entities_didnt_pass_gold_level = 0
         for entity in entities:
-            scorecard = entity.get("scorecards", {}).get(scorecard_name, {})
-            number_of_rules = len(scorecard.get("rules", []))
-            if scorecard.get("level") != "Gold":
-                passed_rules = [rule for rule in scorecard.get("rules", []) if rule.get("status") == "SUCCESS"] or []
-                entities_didnt_pass_gold_level[scorecard.get("level")].append(
+            entity_scorecard_result = entity.get("scorecards", {}).get(scorecard.get("identifier"), {})
+            number_of_rules = len(entity_scorecard_result.get("rules", []))
+            if entity_scorecard_result.get("level") != "Gold":
+                passed_rules = [rule for rule in entity_scorecard_result.get("rules", []) if rule.get("status") == "SUCCESS"] or []
+                entities_didnt_pass_gold_level[entity_scorecard_result.get("level")].append(
                     {
                         "identifier": entity.get("identifier"),
                         "name": entity.get("title"),
@@ -37,7 +37,7 @@ def generate_scorecards_reminders(self,
                     "type": "header",
                     "text": {
                         "type": "plain_text",
-                        "text": f":bulb: {scorecard_name} reminder",
+                        "text": f":bell: {scorecard.get('title')} reminder",
                     }
                 },
                 {

port/client.py

@@ -28,12 +28,16 @@ def search_entities(self, query):
             f"{self.api_url}/v1/entities/search",
             json=query,
             headers=self.headers,
-            params={
-                # "exclude_calculated_properties": "true",
-                # "include": ["blueprint", "identifier"],
-            },
+            params={},
         )
         search_req.raise_for_status()
         return search_req.json()["entities"]
 
+    def get_scorecard(self, blueprint_id: str, scorecard_id: str):
+        scorecard_req = requests.get(
+            f"{self.api_url}/v1/blueprints/{blueprint_id}/scorecards/{scorecard_id}",
+            headers=self.headers,
+        )
+        scorecard_req.raise_for_status()
+        return scorecard_req.json()