2026-01-14 03:09:18 +00:00
|
|
|
# Integration Examples\n\nThis document provides comprehensive examples and patterns for integrating with provisioning APIs, including client libraries, SDKs, error handling\nstrategies, and performance optimization.\n\n## Overview\n\nProvisioning offers multiple integration points:\n\n- REST APIs for workflow management\n- WebSocket APIs for real-time monitoring\n- Configuration APIs for system setup\n- Extension APIs for custom providers and services\n\n## Complete Integration Examples\n\n### Python Integration\n\n#### Full-Featured Python Client\n\n```\nimport asyncio\nimport json\nimport logging\nimport time\nimport requests\nimport websockets\nfrom typing import Dict, List, Optional, Callable\nfrom dataclasses import dataclass\nfrom enum import Enum\n\nclass TaskStatus(Enum):\n PENDING = "Pending"\n RUNNING = "Running"\n COMPLETED = "Completed"\n FAILED = "Failed"\n CANCELLED = "Cancelled"\n\n@dataclass\nclass WorkflowTask:\n id: str\n name: str\n status: TaskStatus\n created_at: str\n started_at: Optional[str] = None\n completed_at: Optional[str] = None\n output: Optional[str] = None\n error: Optional[str] = None\n progress: Optional[float] = None\n\nclass ProvisioningAPIError(Exception):\n """Base exception for provisioning API errors"""\n pass\n\nclass AuthenticationError(ProvisioningAPIError):\n """Authentication failed"""\n pass\n\nclass ValidationError(ProvisioningAPIError):\n """Request validation failed"""\n pass\n\nclass ProvisioningClient:\n """\n Complete Python client for provisioning\n\n Features:\n - REST API integration\n - WebSocket support for real-time updates\n - Automatic token refresh\n - Retry logic with exponential backoff\n - Comprehensive error handling\n """\n\n def __init__(self,\n base_url: str = "http://localhost:9090",\n auth_url: str = "http://localhost:8081",\n username: str = None,\n password: str = None,\n token: str = None):\n self.base_url = base_url\n self.auth_url = auth_url\n self.username = username\n self.password = password\n self.token = token\n self.session = requests.Session()\n self.websocket = None\n self.event_handlers = {}\n\n # Setup logging\n self.logger = logging.getLogger(__name__)\n\n # Configure session with retries\n from requests.adapters import HTTPAdapter\n from urllib3.util.retry import Retry\n\n retry_strategy = Retry(\n total=3,\n status_forcelist=[429, 500, 502, 503, 504],\n method_whitelist=["HEAD", "GET", "OPTIONS"],\n backoff_factor=1\n )\n\n adapter = HTTPAdapter(max_retries=retry_strategy)\n self.session.mount("http://", adapter)\n self.session.mount("https://", adapter)\n\n async def authenticate(self) -> str:\n """Authenticate and get JWT token"""\n if self.token:\n return self.token\n\n if not self.username or not self.password:\n raise AuthenticationError("Username and password required for authentication")\n\n auth_data = {\n "username": self.username,\n "password": self.password\n }\n\n try:\n response = requests.post(f"{self.auth_url}/auth/login", json=auth_data)\n response.raise_for_status()\n\n result = response.json()\n if not result.get('success'):\n raise AuthenticationError(result.get('error', 'Authentication failed'))\n\n self.token = result['data']['token']\n self.session.headers.update({\n 'Authorization': f'Bearer {self.token}'\n })\n\n self.logger.info("Authentication successful")\n return self.token\n\n except requests.RequestException as e:\n raise AuthenticationError(f"Authentication request failed: {e}")\n\n def _make_request(self, method: str, endpoint: str,
|
