finish dagster-pagerduty types (#7668)

python_modules/libraries/dagster-pagerduty/dagster_pagerduty/hooks.py

@@ -1,6 +1,7 @@
 from typing import Callable, Optional
 
 from dagster.core.definitions import OpDefinition, failure_hook
+from dagster.core.definitions.hook_definition import HookDefinition
 from dagster.core.execution.context.hook import HookContext
 
 
@@ -15,15 +16,15 @@ def _dedup_key_fn(context: HookContext) -> str:
     return f"{context.job_name}|{context.op.name}"
 
 
-def _source_fn(context: HookContext):
+def _source_fn(context: HookContext) -> str:
     return f"{context.job_name}"
 
 
 def pagerduty_on_failure(
     severity: str,
     summary_fn: Callable[[HookContext], str] = _default_summary_fn,
     dagit_base_url: Optional[str] = None,
-):
+) -> HookDefinition:
     """Create a hook on step failure events that will trigger a PagerDuty alert.
 
     Args:

python_modules/libraries/dagster-pagerduty/dagster_pagerduty/resources.py

@@ -1,3 +1,5 @@
+from typing import Dict, Optional, cast
+
 import pypd
 
 from dagster import Field, resource
@@ -15,22 +17,22 @@ class PagerDutyService:
     for documentation and more information.
     """
 
-    def __init__(self, routing_key):
+    def __init__(self, routing_key: str):
         self.routing_key = routing_key
 
     def EventV2_create(
         self,
-        summary,
-        source,
-        severity,
-        event_action="trigger",
-        dedup_key=None,
-        timestamp=None,
-        component=None,
-        group=None,
-        event_class=None,
-        custom_details=None,
-    ):
+        summary: str,
+        source: str,
+        severity: str,
+        event_action: str = "trigger",
+        dedup_key: Optional[str] = None,
+        timestamp: Optional[str] = None,
+        component: Optional[str] = None,
+        group: Optional[str] = None,
+        event_class: Optional[str] = None,
+        custom_details: Optional[object] = None,
+    ) -> object:
         """Events API v2 enables you to add PagerDuty's advanced event and incident management
         functionality to any system that can make an outbound HTTP connection.
 
@@ -130,20 +132,22 @@ def EventV2_create(
         if dedup_key is not None:
             data["dedup_key"] = dedup_key
 
+        payload: Dict[str, object] = cast(Dict[str, object], data["payload"])
+
         if timestamp is not None:
-            data["payload"]["timestamp"] = timestamp
+            payload["timestamp"] = timestamp
 
         if component is not None:
-            data["payload"]["component"] = component
+            payload["component"] = component
 
         if group is not None:
-            data["payload"]["group"] = group
+            payload["group"] = group
 
         if event_class is not None:
-            data["payload"]["class"] = event_class
+            payload["class"] = event_class
 
         if custom_details is not None:
-            data["payload"]["custom_details"] = custom_details
+            payload["custom_details"] = custom_details
 
         return pypd.EventV2.create(data=data)