webhook actions via cli

Author: snopoke

taskbadger/cli.py

@@ -41,7 +41,7 @@ def run(
     name: str,
     monitor_id: str = typer.Option(None, help="Associate this task with a monitor."),
     update_frequency: int = typer.Option(5, metavar="SECONDS", min=5, max=300, help="Seconds between updates."),
-    action_def: Tuple[str, integrations.Integrations, str] = typer.Option(
+    action_def: Tuple[str, str, str] = typer.Option(
         (None, None, None),
         "--action",
         "-a",
@@ -63,7 +63,7 @@ def run(
     """
     _configure_api(ctx)
     action = None
-    if all(action_def):
+    if any(action_def):
         trigger, integration, config = action_def
         action = Action(trigger, integrations.from_config(integration, config))
     stale_timeout = update_frequency * 2

taskbadger/integrations.py

@@ -6,15 +6,9 @@
 from taskbadger.internal.models import ActionRequest, ActionRequestConfig
 
 
-class Integrations(str, Enum):
-    email = "email"
-
-
-def from_config(integration: Integrations, config: str):
-    if integration == Integrations.email:
-        split_ = [tuple(item.split(":", 1)) for item in config.split(",")]
-        kwargs = dict(split_)
-        return EmailIntegration(**kwargs)
+def from_config(integration_id: str, config: str):
+    cls = integration_from_id(integration_id)
+    return cls.from_config_string(integration_id, config)
 
 
 class Integration:
@@ -28,6 +22,16 @@ def __post_init__(self):
     def request_config(self):
         raise NotImplementedError
 
+    @classmethod
+    def from_config_string(cls, integration_id, config):
+        kwargs = {"id": integration_id}
+        if config:
+            # convert config string to dict
+            # "to:[email protected],from:[email protected]"
+            split_ = [tuple(item.split(":", 1)) for item in config.split(",")]
+            kwargs.update(dict(split_))
+        return cls(**kwargs)
+
 
 @dataclasses.dataclass
 class Action:
@@ -41,8 +45,8 @@ def to_dict(self) -> Dict[str, Any]:
 @dataclasses.dataclass
 class EmailIntegration(Integration):
     type = "email"
-    id = "email"
     to: str  # custom type
+    id: str = "email"
 
     def request_config(self) -> ActionRequestConfig:
         return ActionRequestConfig.from_dict({"to": self.to})
@@ -55,3 +59,21 @@ class WebhookIntegration(Integration):
 
     def request_config(self) -> ActionRequestConfig:
         return ActionRequestConfig.from_dict({})
+
+
+ALL = [EmailIntegration, WebhookIntegration]
+BY_TYPE = {cls.type: cls for cls in ALL}
+
+
+def integration_from_id(integration_id):
+    type_, _ = get_type_id(integration_id)
+    try:
+        return BY_TYPE[type_]
+    except KeyError:
+        raise TaskbadgerException(f"Unknown integration type: '{type_}'")
+
+
+def get_type_id(integration_id: str):
+    if ":" in integration_id:
+        return integration_id.split(":", 1)
+    return integration_id, None

tests/test_cli_run.py

@@ -53,6 +53,15 @@ def test_cli_run():
     )
 
 
+def test_cli_run_webhook():
+    _test_cli_run(
+        ["echo test"],
+        0,
+        ["task_name", "-a", "cancelled", "webhook:123", ""],
+        action={"trigger": "cancelled", "integration": "webhook:123", "config": {}},
+    )
+
+
 def _test_cli_run(command, return_code, args=None, action=None, update_call_count=1):
     with (
         mock.patch("taskbadger.sdk.task_create.sync_detailed") as create,