Integrations Service

from kubiya import ControlPlaneClient

# Initialize the client
client = ControlPlaneClient(api_key="your-api-key")

# List all integrations
integrations = client.integrations.list()
for integration in integrations:
    print(f"Integration: {integration['name']} ({integration['vendor']}) - Connected: {integration['connected']}")

# Get specific integration details
github_integration = client.integrations.get(integration_id="github-prod")
print(f"GitHub: {github_integration['config']}")

# Get integration credentials
credentials = client.integrations.get_integration_credentials(
    vendor="github",
    id="github-prod"
)

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# List all integrations
integrations = client.integrations.list()

print(f"Total integrations: {len(integrations)}")
for integration in integrations:
    status = "✅" if integration['connected'] else "❌"
    print(f"{status} {integration['name']} ({integration['vendor']})")
    print(f"   Type: {integration['type']}")
    print(f"   Connected: {integration['connected']}")
    if integration.get('last_sync'):
        print(f"   Last Sync: {integration['last_sync']}")
    print("---")

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# Filter to only connected integrations
connected_integrations = client.integrations.list(connected_only=True)

print(f"Connected integrations: {len(connected_integrations)}")
for integration in connected_integrations:
    print(f"✅ {integration['name']} ({integration['vendor']})")

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# Get specific integration by ID
integration = client.integrations.get(integration_id="github-prod")

print(f"Integration: {integration['name']}")
print(f"Vendor: {integration['vendor']}")
print(f"Type: {integration['type']}")
print(f"Connected: {integration['connected']}")
print(f"Configuration: {integration['config']}")

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# Get credentials for GitHub integration
credentials = client.integrations.get_integration_credentials(
    vendor="github",
    id="github-prod"
)

# Use credentials (example: make API call)
token = credentials['token']
import requests
headers = {"Authorization": f"Bearer {token}"}
response = requests.get("https://api.github.com/user/repos", headers=headers)

# Clear sensitive data
token = None
credentials = None

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# Special integrations (Jira, GitHub App) automatically use ID "0"
jira_credentials = client.integrations.get_integration_credentials(
    vendor="jira",
    id="any-id"  # Will be automatically converted to "0"
)

github_app_credentials = client.integrations.get_integration_credentials(
    vendor="github_app",
    id="any-id"  # Will be automatically converted to "0"
)

from kubiya import ControlPlaneClient
from datetime import datetime, timedelta
from typing import Dict, Any

def generate_integration_health_report(client: ControlPlaneClient) -> Dict[str, Any]:
    """Generate comprehensive integration health report."""
    integrations = client.integrations.list()

    connected = []
    disconnected = []
    stale = []
    by_vendor = {}
    by_type = {}

    now = datetime.utcnow()

    for integration in integrations:
        # Track by connection status
        if integration['connected']:
            connected.append(integration)

            # Check for stale sync (no sync in last 24 hours)
            if integration.get('last_sync'):
                last_sync = datetime.fromisoformat(integration['last_sync'].replace('Z', '+00:00'))
                if now - last_sync > timedelta(hours=24):
                    stale.append(integration)
        else:
            disconnected.append(integration)

        # Track by vendor
        vendor = integration['vendor']
        if vendor not in by_vendor:
            by_vendor[vendor] = []
        by_vendor[vendor].append(integration)

        # Track by type
        int_type = integration['type']
        if int_type not in by_type:
            by_type[int_type] = []
        by_type[int_type].append(integration)

    # Print report
    print("Integration Health Report")
    print("=" * 60)
    print(f"Total Integrations: {len(integrations)}")
    print(f"✅ Connected: {len(connected)}")
    print(f"❌ Disconnected: {len(disconnected)}")
    print(f"⚠️  Stale (no sync in 24h): {len(stale)}")

    if disconnected:
        print("\nDisconnected Integrations:")
        for integration in disconnected:
            print(f"  ❌ {integration['name']} ({integration['vendor']})")

    if stale:
        print("\nStale Integrations:")
        for integration in stale:
            hours_since = (now - datetime.fromisoformat(integration['last_sync'].replace('Z', '+00:00'))).seconds // 3600
            print(f"  ⚠️  {integration['name']}: {hours_since}h since last sync")

    print("\nBy Vendor:")
    for vendor, items in by_vendor.items():
        print(f"  {vendor}: {len(items)}")

    print("\nBy Type:")
    for int_type, items in by_type.items():
        print(f"  {int_type}: {len(items)}")

    return {
        "total": len(integrations),
        "connected": len(connected),
        "disconnected": len(disconnected),
        "stale": len(stale),
        "by_vendor": {v: len(items) for v, items in by_vendor.items()},
        "by_type": {t: len(items) for t, items in by_type.items()}
    }

# Usage
client = ControlPlaneClient(api_key="your-api-key")
health_report = generate_integration_health_report(client)

from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import IntegrationError
from typing import List, Dict

def validate_required_integrations(
    client: ControlPlaneClient,
    required: List[str]
) -> Dict[str, Any]:
    """Validate that required integrations are connected."""
    try:
        integrations = client.integrations.list(connected_only=False)

        # Build vendor to integrations mapping
        by_vendor = {}
        for integration in integrations:
            vendor = integration['vendor']
            if vendor not in by_vendor:
                by_vendor[vendor] = []
            by_vendor[vendor].append(integration)

        # Check requirements
        results = {
            "valid": True,
            "present": [],
            "missing": [],
            "disconnected": []
        }

        for vendor in required:
            if vendor in by_vendor:
                # Check if any integration for this vendor is connected
                connected = any(i['connected'] for i in by_vendor[vendor])

                if connected:
                    results["present"].append(vendor)
                else:
                    results["disconnected"].append(vendor)
                    results["valid"] = False
            else:
                results["missing"].append(vendor)
                results["valid"] = False

        # Print report
        if results["valid"]:
            print(f"✅ All {len(required)} required integrations are connected")
        else:
            if results["missing"]:
                print(f"❌ Missing {len(results['missing'])} integrations:")
                for vendor in results["missing"]:
                    print(f"  - {vendor}")

            if results["disconnected"]:
                print(f"⚠️  {len(results['disconnected'])} integrations disconnected:")
                for vendor in results["disconnected"]:
                    print(f"  - {vendor}")

        return results

    except IntegrationError as e:
        print(f"Failed to validate integrations: {e}")
        return None

# Usage
client = ControlPlaneClient(api_key="your-api-key")

required_integrations = ["github", "slack", "aws", "datadog"]
validation = validate_required_integrations(client, required_integrations)

if not validation['valid']:
    print("\nPlease configure missing integrations in Kubiya Dashboard")

from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import IntegrationError
from contextlib import contextmanager
import gc

@contextmanager
def integration_credentials(client: ControlPlaneClient, vendor: str, integration_id: str):
    """Context manager for safe credential access."""
    credentials = None
    try:
        credentials = client.integrations.get_integration_credentials(
            vendor=vendor,
            id=integration_id
        )
        yield credentials

    except IntegrationError as e:
        print(f"Failed to get credentials for {vendor}/{integration_id}: {e}")
        raise

    finally:
        if credentials:
            credentials = None
        gc.collect()

# Usage
client = ControlPlaneClient(api_key="your-api-key")

with integration_credentials(client, "github", "github-prod") as creds:
    token = creds['token']

    # Use credentials
    import requests
    headers = {"Authorization": f"Bearer {token}"}
    response = requests.get("https://api.github.com/user/repos", headers=headers)

    print(f"Found {len(response.json())} repositories")

# Credentials automatically cleared from memory

from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import IntegrationError
from typing import Dict, Any, List
import concurrent.futures

def query_multiple_integrations(
    client: ControlPlaneClient,
    queries: Dict[str, callable]
) -> Dict[str, Any]:
    """Execute queries across multiple integrations in parallel."""

    results = {}

    def execute_query(vendor_id: str, query_func: callable):
        """Execute single integration query."""
        try:
            return vendor_id, query_func(client)
        except IntegrationError as e:
            return vendor_id, {"error": str(e)}

    # Execute queries in parallel
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        futures = [
            executor.submit(execute_query, vendor_id, query_func)
            for vendor_id, query_func in queries.items()
        ]

        for future in concurrent.futures.as_completed(futures):
            vendor_id, result = future.result()
            results[vendor_id] = result

    return results

# Define integration-specific queries
def query_github(client: ControlPlaneClient) -> Dict:
    """Query GitHub integration."""
    creds = client.integrations.get_integration_credentials(vendor="github", id="github-prod")
    # Make API call using credentials
    return {"repos": 42, "status": "active"}

def query_aws(client: ControlPlaneClient) -> Dict:
    """Query AWS integration."""
    creds = client.integrations.get_integration_credentials(vendor="aws", id="aws-prod")
    # Make API call using credentials
    return {"instances": 15, "status": "active"}

def query_slack(client: ControlPlaneClient) -> Dict:
    """Query Slack integration."""
    creds = client.integrations.get_integration_credentials(vendor="slack", id="slack-main")
    # Make API call using credentials
    return {"channels": 23, "status": "active"}

# Usage
client = ControlPlaneClient(api_key="your-api-key")

queries = {
    "github": query_github,
    "aws": query_aws,
    "slack": query_slack
}

results = query_multiple_integrations(client, queries)

for vendor, result in results.items():
    if "error" in result:
        print(f"❌ {vendor}: {result['error']}")
    else:
        print(f"✅ {vendor}: {result}")

from kubiya import ControlPlaneClient
from datetime import datetime
from collections import Counter

def analyze_integration_usage(client: ControlPlaneClient) -> dict:
    """Analyze integration usage patterns."""
    integrations = client.integrations.list()

    # Categorize integrations
    vendor_counts = Counter(i['vendor'] for i in integrations)
    type_counts = Counter(i['type'] for i in integrations)
    connection_rate = sum(1 for i in integrations if i['connected']) / len(integrations)

    # Find most/least used types
    most_common_vendor = vendor_counts.most_common(1)[0] if vendor_counts else None
    most_common_type = type_counts.most_common(1)[0] if type_counts else None

    print("Integration Usage Analytics")
    print("=" * 60)
    print(f"Total Integrations: {len(integrations)}")
    print(f"Connection Rate: {connection_rate * 100:.1f}%")

    if most_common_vendor:
        print(f"Most Used Vendor: {most_common_vendor[0]} ({most_common_vendor[1]} integrations)")

    if most_common_type:
        print(f"Most Used Type: {most_common_type[0]} ({most_common_type[1]} integrations)")

    print("\nVendor Distribution:")
    for vendor, count in vendor_counts.most_common():
        print(f"  {vendor}: {count}")

    print("\nType Distribution:")
    for int_type, count in type_counts.most_common():
        print(f"  {int_type}: {count}")

    return {
        "total": len(integrations),
        "connection_rate": connection_rate,
        "vendor_counts": dict(vendor_counts),
        "type_counts": dict(type_counts)
    }

# Usage
client = ControlPlaneClient(api_key="your-api-key")
usage_report = analyze_integration_usage(client)

from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import IntegrationError
from kubiya.core.exceptions import (
    APIError as KubiyaAPIError,
    AuthenticationError as KubiyaAuthenticationError,
    TimeoutError as KubiyaTimeoutError
)

client = ControlPlaneClient(api_key="your-api-key")

# Handle integration errors
try:
    integrations = client.integrations.list()
except IntegrationError as e:
    print(f"Integration operation failed: {e}")
except KubiyaAuthenticationError as e:
    print(f"Authentication failed: {e}")
except KubiyaAPIError as e:
    print(f"API error: {e.status_code} - {e.message}")

# Handle integration not found
try:
    integration = client.integrations.get(integration_id="non-existent")
except IntegrationError as e:
    if "not found" in str(e).lower():
        print("Integration not found")
    else:
        print(f"Error getting integration: {e}")

# Handle credential access errors
try:
    credentials = client.integrations.get_integration_credentials(
        vendor="github",
        id="invalid-id"
    )
except IntegrationError as e:
    if "not found" in str(e).lower():
        print("Integration credentials not found")
    elif "permission denied" in str(e).lower():
        print("Insufficient permissions")
    else:
        print(f"Failed to get credentials: {e}")

from kubiya import ControlPlaneClient

def get_integration_safely(client: ControlPlaneClient, integration_id: str) -> dict:
    """Get integration and verify it's connected."""
    integration = client.integrations.get(integration_id=integration_id)

    if not integration['connected']:
        raise ValueError(f"Integration {integration_id} is not connected")

    return integration

# Usage
client = ControlPlaneClient(api_key="your-api-key")
try:
    github = get_integration_safely(client, "github-prod")
    # Safe to use
except ValueError as e:
    print(f"Cannot use integration: {e}")

from kubiya import ControlPlaneClient
from datetime import datetime, timedelta

class IntegrationCache:
    """Cache integration list to reduce API calls."""

    def __init__(self, client: ControlPlaneClient, ttl_seconds: int = 300):
        self.client = client
        self.ttl_seconds = ttl_seconds
        self._cache = None
        self._expiry = None

    def get_integrations(self) -> list:
        """Get integrations with caching."""
        now = datetime.utcnow()

        # Check cache
        if self._cache and self._expiry and now < self._expiry:
            return self._cache

        # Fetch and cache
        self._cache = self.client.integrations.list()
        self._expiry = now + timedelta(seconds=self.ttl_seconds)

        return self._cache

    def invalidate(self):
        """Clear cache."""
        self._cache = None
        self._expiry = None

# Usage
cache = IntegrationCache(client, ttl_seconds=300)
integrations = cache.get_integrations()

from kubiya import ControlPlaneClient

# Special integrations that always use ID "0"
SPECIAL_INTEGRATIONS = {"jira", "github_app"}

def get_credentials_safely(
    client: ControlPlaneClient,
    vendor: str,
    integration_id: str
):
    """Get credentials with special integration handling."""
    # SDK handles this automatically, but good to be aware
    if vendor.lower() in SPECIAL_INTEGRATIONS:
        print(f"Note: {vendor} is a special integration using ID '0'")

    return client.integrations.get_integration_credentials(
        vendor=vendor,
        id=integration_id
    )

# Usage
client = ControlPlaneClient(api_key="your-api-key")
jira_creds = get_credentials_safely(client, "jira", "any-id")

import schedule
from kubiya import ControlPlaneClient

def check_critical_integrations(client: ControlPlaneClient):
    """Monitor critical integrations."""
    critical = ["github", "slack", "aws"]

    connected = client.integrations.list(connected_only=True)
    connected_vendors = {i['vendor'] for i in connected}

    for vendor in critical:
        if vendor not in connected_vendors:
            # Alert or take action
            print(f"ALERT: Critical integration {vendor} is not connected!")

# Schedule checks
client = ControlPlaneClient(api_key="your-api-key")
schedule.every(15).minutes.do(lambda: check_critical_integrations(client))

{
    "id": "github-prod",
    "name": "GitHub Production",
    "vendor": "github",
    "type": "version_control",  # version_control, communication, cloud_provider, ticketing, etc.
    "connected": true,
    "config": {
        "organization": "acme-corp",
        "repos": ["backend", "frontend"]
    },
    "created_at": "2025-10-15T09:00:00Z",
    "last_sync": "2025-12-11T10:25:00Z"
}