Improved type safety and better variable naming

Author: Kamlesh72

backend/analytics_server/mhq/api/incidents.py

@@ -1,5 +1,5 @@
 import json
-from typing import Dict, List
+from typing import Dict, List, Optional as typeOptional
 
 from datetime import datetime
 
@@ -41,7 +41,10 @@
     ),
 )
 def get_resolved_incidents(
-    team_id: str, from_time: datetime, to_time: datetime, pr_filter: dict = None
+    team_id: str,
+    from_time: datetime,
+    to_time: datetime,
+    pr_filter: typeOptional[Dict] = None,
 ):
 
     query_validator = get_query_validator()
@@ -78,7 +81,7 @@ def get_deployments_with_related_incidents(
     team_id: str,
     from_time: datetime,
     to_time: datetime,
-    pr_filter: dict = None,
+    pr_filter: typeOptional[Dict] = None,
     workflow_filter: WorkflowFilter = None,
 ):
     query_validator = get_query_validator()
@@ -126,7 +129,10 @@ def get_deployments_with_related_incidents(
     ),
 )
 def get_team_mttr(
-    team_id: str, from_time: datetime, to_time: datetime, pr_filter: dict = None
+    team_id: str,
+    from_time: datetime,
+    to_time: datetime,
+    pr_filter: typeOptional[Dict] = None,
 ):
     query_validator = get_query_validator()
     interval = query_validator.interval_validator(from_time, to_time)
@@ -156,7 +162,10 @@ def get_team_mttr(
     ),
 )
 def get_team_mttr_trends(
-    team_id: str, from_time: datetime, to_time: datetime, pr_filter: dict = None
+    team_id: str,
+    from_time: datetime,
+    to_time: datetime,
+    pr_filter: typeOptional[Dict] = None,
 ):
     query_validator = get_query_validator()
     interval = query_validator.interval_validator(from_time, to_time)
@@ -197,7 +206,7 @@ def get_team_cfr(
     team_id: str,
     from_time: datetime,
     to_time: datetime,
-    pr_filter: dict = None,
+    pr_filter: typeOptional[Dict] = None,
     workflow_filter: WorkflowFilter = None,
 ):
 
@@ -243,7 +252,7 @@ def get_team_cfr_trends(
     team_id: str,
     from_time: datetime,
     to_time: datetime,
-    pr_filter: dict = None,
+    pr_filter: typeOptional[Dict] = None,
     workflow_filter: WorkflowFilter = None,
 ):
 

backend/analytics_server/mhq/service/incidents/incidents.py

@@ -28,7 +28,7 @@
     get_settings_service,
 )
 from mhq.store.repos.incidents import IncidentsRepoService
-from mhq.service.incidents.models.adapter import IncidentPRAdapter
+from mhq.service.incidents.models.adapter import adaptIncidentPR
 from mhq.service.code.pr_filter import apply_pr_filter
 from dataclasses import asdict
 
@@ -116,19 +116,26 @@ def get_team_pr_incidents(
             from_time=interval.from_time, to_time=time_now()
         )
         resolution_prs_filter: PRFilter = apply_pr_filter(
-            asdict(pr_filter),
-            EntityType.TEAM,
-            team_id,
-            [SettingType.EXCLUDED_PRS_SETTING, SettingType.INCIDENT_PRS_SETTING],
+            pr_filter=asdict(pr_filter),
+            entity_type=EntityType.TEAM,
+            entity_id=team_id,
+            setting_types=[
+                SettingType.EXCLUDED_PRS_SETTING,
+                SettingType.INCIDENT_PRS_SETTING,
+            ],
         )
 
         resolution_prs = self._code_repo_service.get_prs_merged_in_interval(
-            team_repo_ids, resolution_prs_interval, resolution_prs_filter
+            repo_ids=team_repo_ids,
+            interval=resolution_prs_interval,
+            pr_filter=resolution_prs_filter,
         )
 
         pr_numbers: List[str] = []
         pr_incidents: List[Incident] = []
-        repo_id_to_pr_number_to_pr_map: Dict[str, Dict[str, PullRequest]] = {}
+        repo_id_to_pr_number_to_pr_map: Dict[str, Dict[str, PullRequest]] = defaultdict(
+            dict
+        )
 
         for pr in resolution_prs:
             for filter in incident_prs_setting.filters:
@@ -138,26 +145,24 @@ def get_team_pr_incidents(
 
                 if incident_pr_number:
                     pr_numbers.append(incident_pr_number)
-                    if str(pr.repo_id) not in repo_id_to_pr_number_to_pr_map:
-                        repo_id_to_pr_number_to_pr_map[str(pr.repo_id)] = {}
                     repo_id_to_pr_number_to_pr_map[str(pr.repo_id)][
                         incident_pr_number
                     ] = pr
                     break
 
         prs = self._code_repo_service.get_prs_merged_in_interval_by_numbers(
-            list(repo_id_to_pr_number_to_pr_map.keys()), interval, pr_numbers, pr_filter
+            repo_ids=list(repo_id_to_pr_number_to_pr_map.keys()),
+            interval=interval,
+            numbers=pr_numbers,
+            pr_filter=pr_filter,
         )
 
         for pr in prs:
-            if (
-                str(pr.repo_id) not in repo_id_to_pr_number_to_pr_map
-                or pr.number not in repo_id_to_pr_number_to_pr_map[str(pr.repo_id)]
-            ):
+            if pr.number not in repo_id_to_pr_number_to_pr_map[str(pr.repo_id)]:
                 continue
 
             resolution_pr = repo_id_to_pr_number_to_pr_map[str(pr.repo_id)][pr.number]
-            adapted_pr_incident = IncidentPRAdapter.adapt(pr, resolution_pr)
+            adapted_pr_incident = adaptIncidentPR(pr, resolution_pr)
             pr_incidents.append(adapted_pr_incident)
 
         return pr_incidents

backend/analytics_server/mhq/service/incidents/models/adapter.py

@@ -3,22 +3,20 @@
 from mhq.store.models.incidents.enums import IncidentStatus, IncidentType
 
 
-class IncidentPRAdapter:
-    @staticmethod
-    def adapt(incident_pr: PullRequest, resolution_pr: PullRequest) -> Incident:
-        return Incident(
-            id=incident_pr.id,
-            provider=incident_pr.provider,
-            key=str(incident_pr.id),
-            title=incident_pr.title,
-            incident_number=int(incident_pr.number),
-            status=IncidentStatus.RESOLVED.value,
-            creation_date=incident_pr.state_changed_at,
-            acknowledged_date=resolution_pr.created_at,
-            resolved_date=resolution_pr.state_changed_at,
-            assigned_to=resolution_pr.author,
-            assignees=[resolution_pr.author],
-            url=incident_pr.url,
-            meta={},
-            incident_type=IncidentType.REVERT_PR,
-        )
+def adaptIncidentPR(incident_pr: PullRequest, resolution_pr: PullRequest) -> Incident:
+    return Incident(
+        id=incident_pr.id,
+        provider=incident_pr.provider,
+        key=str(incident_pr.id),
+        title=incident_pr.title,
+        incident_number=int(incident_pr.number),
+        status=IncidentStatus.RESOLVED.value,
+        creation_date=incident_pr.state_changed_at,
+        acknowledged_date=resolution_pr.created_at,
+        resolved_date=resolution_pr.state_changed_at,
+        assigned_to=resolution_pr.author,
+        assignees=[resolution_pr.author],
+        url=incident_pr.url,
+        meta={},
+        incident_type=IncidentType.REVERT_PR,
+    )

backend/analytics_server/mhq/service/settings/models.py

@@ -1,6 +1,6 @@
 from dataclasses import dataclass
 from datetime import datetime
-from typing import List
+from typing import List, TypedDict, Literal
 
 from mhq.store.models import EntityType
 from mhq.store.models.incidents.enums import IncidentSource, IncidentType
@@ -46,9 +46,8 @@ class DefaultSyncDaysSetting(BaseSetting):
     default_sync_days: int
 
 
-@dataclass
-class IncidentPRFilter:
-    field: str
+class IncidentPRFilter(TypedDict):
+    field: Literal["title", "head_branch"]
     value: str
 
 

backend/analytics_server/mhq/store/models/code/filter.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import List, Dict
+from typing import List, Dict, Optional
 
 from sqlalchemy import and_, or_
 
@@ -13,7 +13,7 @@ class PRFilter:
     repo_filters: Dict[str, Dict] = None
     excluded_pr_ids: List[str] = None
     max_cycle_time: int = None
-    incident_pr_filters: List[Dict] = None
+    incident_pr_filters: Optional[List[Dict]] = None
 
     class RepoFilter:
         def __init__(self, repo_id: str, repo_filters=None):

backend/analytics_server/tests/service/Incidents/test_team_pr_incidents.py

@@ -56,8 +56,8 @@ def get_settings_map(self, *args, **kwargs):
 
 
 class FakeCodeRepoService:
-    def __init__(self, prs_using_filters, prs_using_numbers):
-        self._prs_using_filters = prs_using_filters
+    def __init__(self, resolution_prs, prs_using_numbers):
+        self._resolution_prs = resolution_prs
         self._prs_using_numbers = prs_using_numbers
 
     def get_active_team_repos_by_team_id(self, *args, **kwargs):
@@ -73,7 +73,7 @@ def get_active_team_repos_by_team_id(self, *args, **kwargs):
         ]
 
     def get_prs_merged_in_interval(self, *args, **kwargs):
-        return self._prs_using_filters
+        return self._resolution_prs
 
     def get_prs_merged_in_interval_by_numbers(self, *args, **kwargs):
         return self._prs_using_numbers
@@ -103,7 +103,7 @@ def test_get_team_pr_incidents_no_filters():
 
 def test_get_team_pr_incidents_with_filters():
 
-    prs_using_filters = [
+    resolution_prs = [
         PullRequest(
             id="pr_2_of_repo_1",
             repo_id="repo_1",
@@ -134,7 +134,7 @@ def test_get_team_pr_incidents_with_filters():
     incident_service = IncidentService(
         FakeIncidentsRepoService(),
         FakeSettingsService(),
-        FakeCodeRepoService(prs_using_filters, prs_using_numbers),
+        FakeCodeRepoService(resolution_prs, prs_using_numbers),
     )
 
     expected_result_keys = [
@@ -153,7 +153,7 @@ def test_get_team_pr_incidents_with_filters():
 
 def test_get_team_pr_incidents_with_multiple_repos_but_no_incidents():
 
-    prs_using_filters = [
+    resolution_prs = [
         PullRequest(
             id="pr_2_of_repo_1",
             repo_id="repo_1",
@@ -173,7 +173,7 @@ def test_get_team_pr_incidents_with_multiple_repos_but_no_incidents():
     incident_service = IncidentService(
         FakeIncidentsRepoService(),
         FakeSettingsService(),
-        FakeCodeRepoService(prs_using_filters, prs_using_numbers),
+        FakeCodeRepoService(resolution_prs, prs_using_numbers),
     )
 
     expected_result_keys = []
@@ -189,7 +189,7 @@ def test_get_team_pr_incidents_with_multiple_repos_but_no_incidents():
 
 def test_get_team_pr_incidents_with_multiple_repos_and_incidents():
 
-    prs_using_filters = [
+    resolution_prs = [
         PullRequest(
             id="pr_2_of_repo_1", repo_id="repo_1", number="2", head_branch="revert-1"
         ),
@@ -219,7 +219,7 @@ def test_get_team_pr_incidents_with_multiple_repos_and_incidents():
     incident_service = IncidentService(
         FakeIncidentsRepoService(),
         FakeSettingsService(),
-        FakeCodeRepoService(prs_using_filters, prs_using_numbers),
+        FakeCodeRepoService(resolution_prs, prs_using_numbers),
     )
 
     expected_result_keys = [