Adding the integration commands

Author: armiiller

api.py

@@ -216,6 +216,36 @@ def delete_broadcast(self, broadcast_id: str) -> Dict[str, Any]:
         response.raise_for_status()
         return response.json() if response.content else {"message": "Broadcast deleted successfully"}
 
+    # INTEGRATIONS
+    # ============
+    def list_integrations(self, limit: int = 10, offset: int = 0, search: Optional[str] = None, enabled: Optional[bool] = None) -> Dict[str, Any]:
+        """List all integrations in PagerTree."""
+        params = {k: v for k, v in {"limit": limit, "offset": offset, "q": search, "enabled": enabled}.items() if v is not None}
+        response = self.session.get(f"{self.base_url}/integrations", params=params)
+        response.raise_for_status()
+        data = response.json()
+        return {
+            "data": data.get("data", []),
+            "total": data.get("total_count", 0),
+            "has_more": data.get("has_more", False),
+            "limit": limit,
+            "offset": offset
+        }
+
+    def show_integration(self, integration_id: str) -> Dict[str, Any]:
+        """Fetch a single integration by ID from PagerTree."""
+        response = self.session.get(f"{self.base_url}/integrations/{integration_id}")
+        response.raise_for_status()
+        return response.json()
+
+    def update_integration(self, integration_id: str, enabled: Optional[bool] = None) -> Dict[str, Any]:
+        """Enable an integration in PagerTree."""
+        payload = {"enabled": enabled}
+        payload = {k: v for k, v in payload.items() if v is not None}
+        response = self.session.put(f"{self.base_url}/integrations/{integration_id}", json=payload)
+        response.raise_for_status()
+        return response.json()
+        
     # TEAMS
     # ======
 

commands/integrations.py

@@ -0,0 +1,85 @@
+import click
+import json
+from utils import display_paginated_results, handle_api_error, format_item_details
+
+@click.group()
+def integrations():
+    """Commands for managing PagerTree integrations."""
+    pass
+
+@integrations.command(name="show")
+@click.argument("integration_id", required=True)
+@click.pass_context
+def show_integration_cmd(ctx, integration_id):
+    """Show details of a specific integration in PagerTree."""
+    try:
+        client = ctx.obj.client  # Get PagerTreeClient from context
+        integration = client.show_integration(integration_id)
+        fields = {
+            "id": "ID",
+            "name": "Name",
+            "integration_type.name": "Type",
+            "enabled": "Enabled",
+            "created_at": "Created At",
+            "updated_at": "Updated At"
+        }
+        format_item_details(integration, fields)
+    except Exception as e:
+        handle_api_error(e, "showing integration")
+
+@integrations.command(name="list")
+@click.option("--limit", default=10, type=click.IntRange(1, 100), help="Number of integrations per page")
+@click.option("--offset", default=0, type=click.IntRange(0), help="Starting point for pagination")
+@click.option("--search", help="Search for integrations by name or type")
+@click.option("--enabled", is_flag=True, help="Filter for enabled integrations", default=None)
+@click.option("--disabled", is_flag=True, help="Filter for disabled integrations")
+@click.pass_context
+def list_integrations_cmd(ctx, limit, offset, search, enabled, disabled):
+    """List integrations in PagerTree with pagination."""
+    try:
+        # Ensure --enabled and --disabled are mutually exclusive
+        if enabled and disabled:
+            click.echo("Error: --enabled and --disabled cannot be used together.")
+            return
+
+        # Set enabled parameter: True for --enabled, False for --disabled, None if neither
+        enabled_param = 1 if enabled else 0 if disabled else None
+
+        client = ctx.obj.client  # Get PagerTreeClient from context
+        logger = ctx.obj.logger  # Get logger from context
+        logger.debug(f"Listing integrations with limit={limit}, offset={offset}, search={search}, enabled={enabled_param}")
+        result = client.list_integrations(limit=limit, offset=offset, search=search, enabled=enabled_param)
+        logger.debug(f"Full response: {json.dumps(result, indent=2)}")
+        integrations_list = result["data"]
+        total = result["total"]
+        # Prepare table data
+        headers = ["ID", "Name", "Type", "Enabled"]
+        table_data = [[integration.get("id"), integration.get("name"), integration.get("integration_type").get("name"), integration.get("enabled")] for integration in integrations_list]
+        display_paginated_results(integrations_list, total, limit, offset, "integration", headers, table_data)
+    except Exception as e:
+        logger.error(f"Error listing integrations: {str(e)}")
+        handle_api_error(e, action="listing integrations")
+
+@integrations.command(name="enable")
+@click.argument("integration_id", required=True)
+@click.pass_context
+def enable_integration_cmd(ctx, integration_id):
+    """Enable an integration in PagerTree."""
+    try:
+        client = ctx.obj.client  # Get PagerTreeClient from context
+        result = client.update_integration(integration_id, enabled=True)
+        click.echo(f"Integration enabled successfully: {result.get('id')}")
+    except Exception as e:
+        handle_api_error(e, action="enabling integration")
+
+@integrations.command(name="disable")
+@click.argument("integration_id", required=True)
+@click.pass_context
+def disable_integration_cmd(ctx, integration_id):
+    """Disable an integration in PagerTree."""
+    try:
+        client = ctx.obj.client  # Get PagerTreeClient from context
+        result = client.update_integration(integration_id, enabled=False)
+        click.echo(f"Integration disabled successfully: {result.get('id')}")
+    except Exception as e:
+        handle_api_error(e, action="disabling integration")

requirements.txt

@@ -3,7 +3,9 @@ certifi==2025.1.31
 charset-normalizer==3.4.1
 click==8.1.8
 idna==3.10
+jsonpath-ng==1.7.0
 packaging==24.2
+ply==3.11
 pyinstaller==6.12.0
 pyinstaller-hooks-contrib==2025.2
 python-dotenv==1.1.0

utils.py

@@ -1,5 +1,8 @@
 import click
 import requests
+from typing import Dict, Any
+from jsonpath_ng import parse
+from jsonpath_ng.exceptions import JsonPathParserError, JsonPathLexerError
 from tabulate import tabulate
 
 def handle_api_error(e, action="performing action"):
@@ -23,14 +26,33 @@ def display_paginated_results(items, total, limit, offset, item_type="item", tab
     if offset + limit < total:
         click.echo(f"More {item_type}s available. Use --offset {offset + limit} to see next page.")
 
-def format_item_details(item, fields):
-    """Format and display item details as a table using tabulate."""
+def format_item_details(item: Dict[str, Any], fields: Dict[str, str]) -> None:
+    """Format and display item details as a table using tabulate, supporting JSON path notation."""
     # Prepare table data: each row is [Display Name, Value]
-    table_data = [
-        [display_name, item.get(field_name, "N/A")]
-        for field_name, display_name in fields.items()
-    ]
+    table_data = []
+    for field_path, display_name in fields.items():
+        try:
+            # Try to parse the field_path as a JSON path
+            jsonpath_expr = parse(field_path)
+            matches = jsonpath_expr.find(item)
+            value = matches[0].value if matches else "N/A"
+        except (JsonPathParserError, JsonPathLexerError, IndexError, KeyError):
+            # Fallback to direct dictionary access for simple field names
+            value = item.get(field_path, "N/A")
+
+        # Format specific types for better readability
+        if isinstance(value, bool):
+            value = "Yes" if value else "No"
+        elif isinstance(value, list):
+            value = ", ".join(str(v) for v in value) if value else "None"
+        elif isinstance(value, dict):
+            value = str(value)  # Convert dict to string for simplicity
+        elif value is None:
+            value = "N/A"
+        
+        table_data.append([display_name, value])
+    
     # Define table headers
     headers = ["Field", "Value"]
-    # Display the table using tabulate with github format
+    # Display the table using tabulate with simple format
     click.echo(tabulate(table_data, headers=headers, tablefmt="simple", maxcolwidths=[None, 50]))