#!/usr/bin/env python3 """ Web of Facts Alert API Client - Python Example This example demonstrates how to create, update, and query alerts using the Web of Facts Alert API. Requirements: pip install requests Usage: 1. Update API_URL and API_KEY below 2. Run: python python_alerts.py """ import requests import json from datetime import datetime, timedelta from typing import Optional, Dict, Any, List class WebOfFactsAlerts: """Client for the Web of Facts Alert API.""" def __init__(self, api_url: str, api_key: str): """ Initialize the alert client. Args: api_url: Base URL for the alerts API (e.g., https://example.com/api/external/alerts) api_key: Your Web of Facts API key """ self.api_url = api_url.rstrip('/') self.headers = { 'X-API-Key': api_key, 'Content-Type': 'application/json' } def create_alert( self, external_id: str, link_to: str, name: Optional[str] = None, severity: str = 'warning', message: str = '', source: str = 'python-client', notes: str = '' ) -> Dict[str, Any]: """ Create a new alert linked to an entity. Args: external_id: Unique identifier for this alert in your system link_to: Entity to link to (external_id, node id, or label) name: Display name (defaults to "Alert: {external_id}") severity: One of: info, warning, error, critical message: Detailed alert message source: Source system name notes: Additional notes Returns: API response as dictionary Raises: requests.HTTPError: If the request fails """ payload = { 'external_id': external_id, 'link_to': link_to, 'state': 'active', 'severity': severity, 'message': message, 'source': source, 'notes': notes } if name: payload['name'] = name response = requests.post(self.api_url, json=payload, headers=self.headers) response.raise_for_status() return response.json() def update_alert( self, external_id: str, link_to: str, state: str, message: str = '', acknowledged_by: Optional[str] = None ) -> Dict[str, Any]: """ Update an existing alert's state. Args: external_id: The alert's external identifier link_to: Entity the alert is linked to state: New state (active, acknowledged, cleared) message: Updated message acknowledged_by: Who acknowledged (for acknowledged state) Returns: API response as dictionary """ payload = { 'external_id': external_id, 'link_to': link_to, 'state': state, 'message': message } if acknowledged_by and state == 'acknowledged': payload['acknowledged_by'] = acknowledged_by response = requests.put(self.api_url, json=payload, headers=self.headers) response.raise_for_status() return response.json() def acknowledge_alert( self, external_id: str, link_to: str, acknowledged_by: str = 'api-user', message: str = '' ) -> Dict[str, Any]: """ Acknowledge an alert. Args: external_id: The alert's external identifier link_to: Entity the alert is linked to acknowledged_by: Who is acknowledging message: Optional acknowledgment message Returns: API response as dictionary """ return self.update_alert( external_id=external_id, link_to=link_to, state='acknowledged', message=message, acknowledged_by=acknowledged_by ) def clear_alert( self, external_id: str, link_to: str, message: str = '' ) -> Dict[str, Any]: """ Clear/resolve an alert. Args: external_id: The alert's external identifier link_to: Entity the alert is linked to message: Resolution message Returns: API response as dictionary """ return self.update_alert( external_id=external_id, link_to=link_to, state='cleared', message=message ) def get_alerts( self, state: Optional[str] = None, external_id: Optional[str] = None, since: Optional[datetime] = None, limit: int = 100 ) -> List[Dict[str, Any]]: """ Query alerts from the project. Args: state: Filter by state (active, acknowledged, cleared) external_id: Get specific alert by external_id since: Only return alerts updated after this time limit: Maximum number of results (max 500) Returns: List of alert objects """ params = {'limit': min(limit, 500)} if state: params['state'] = state if external_id: params['external_id'] = external_id if since: params['since'] = since.isoformat() response = requests.get(self.api_url, params=params, headers=self.headers) response.raise_for_status() data = response.json() return data.get('data', {}).get('alerts', []) def get_active_alerts(self) -> List[Dict[str, Any]]: """Get all active (firing) alerts.""" return self.get_alerts(state='active') def get_recent_alerts(self, hours: int = 24) -> List[Dict[str, Any]]: """Get alerts updated in the last N hours.""" since = datetime.utcnow() - timedelta(hours=hours) return self.get_alerts(since=since) def main(): """Example usage of the WebOfFactsAlerts client.""" # Configuration - UPDATE THESE VALUES API_URL = 'https://your-domain.com/api/external/alerts' API_KEY = 'wof_xxxxxxxx_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx' # Entity to link alerts to - must exist in your project # Can be: external_id, internal node id, or exact label TARGET_ENTITY = 'server-01' # Initialize client client = WebOfFactsAlerts(API_URL, API_KEY) print("=== Web of Facts Alert API - Python Example ===\n") # Example 1: Create an alert print("1. Creating a new alert...") try: result = client.create_alert( external_id='python-demo-001', link_to=TARGET_ENTITY, name='High Memory Usage', severity='warning', message='Memory usage exceeded 85% threshold', source='python-demo' ) print(f" Created alert ID: {result['data']['id']}") print(f" Linked to: {result['data']['linked_to']['label']}") except requests.HTTPError as e: print(f" Error: {e.response.status_code} - {e.response.text}") # Example 2: Query active alerts print("\n2. Querying active alerts...") try: alerts = client.get_active_alerts() print(f" Found {len(alerts)} active alert(s)") for alert in alerts[:3]: # Show first 3 data = alert.get('data', {}) print(f" - {alert['label']} ({data.get('severity', 'unknown')})") except requests.HTTPError as e: print(f" Error: {e.response.status_code} - {e.response.text}") # Example 3: Acknowledge an alert print("\n3. Acknowledging alert...") try: result = client.acknowledge_alert( external_id='python-demo-001', link_to=TARGET_ENTITY, acknowledged_by='demo-user', message='Looking into this issue' ) print(f" Alert state: {result['data']['state']}") except requests.HTTPError as e: print(f" Error: {e.response.status_code} - {e.response.text}") # Example 4: Clear the alert print("\n4. Clearing alert...") try: result = client.clear_alert( external_id='python-demo-001', link_to=TARGET_ENTITY, message='Issue resolved - memory leak fixed' ) print(f" Alert state: {result['data']['state']}") except requests.HTTPError as e: print(f" Error: {e.response.status_code} - {e.response.text}") # Example 5: Get recent alerts print("\n5. Getting alerts from last 24 hours...") try: alerts = client.get_recent_alerts(hours=24) print(f" Found {len(alerts)} alert(s) in last 24 hours") except requests.HTTPError as e: print(f" Error: {e.response.status_code} - {e.response.text}") print("\n=== Done ===") if __name__ == '__main__': main()