diff --git a/sdk/python/requirements.txt b/sdk/python/requirements.txt index 9d98daa..0032d33 100644 --- a/sdk/python/requirements.txt +++ b/sdk/python/requirements.txt @@ -4,3 +4,4 @@ aiohttp>=3.10.0 websockets>=14.0.0 pydantic>=2.10.0 typing-extensions>=4.12.0 +pyjwt>=2.10.0 diff --git a/sdk/python/telecom_sdk/__init__.py b/sdk/python/telecom_sdk/__init__.py index 97b3a2d..26a788e 100644 --- a/sdk/python/telecom_sdk/__init__.py +++ b/sdk/python/telecom_sdk/__init__.py @@ -1,4 +1,15 @@ from .client import TelecomSDK +from .auth import AuthProvider +from .api import ( + APIClient, + SubscriberAPI, + UsageAPI, + PaymentAPI, + RatingPlanAPI, + SystemAPI, + GraphQLAPI, +) +from .websocket import WebSocketClient from .types import ( Subscriber, UsageStats, @@ -6,25 +17,54 @@ from .types import ( SystemStats, HealthStatus, WebSocketMessage, + UsageEvent, + RatingPlan, + RealTimeUsage, + PaginatedResponse, ) from .exceptions import ( TelecomError, AuthenticationError, APIError, NetworkError, + ValidationError, + RateLimitError, + ServerError, ) __version__ = "1.0.0" __all__ = [ + # Main SDK client "TelecomSDK", + # Authentication + "AuthProvider", + # API modules + "APIClient", + "SubscriberAPI", + "UsageAPI", + "PaymentAPI", + "RatingPlanAPI", + "SystemAPI", + "GraphQLAPI", + # WebSocket + "WebSocketClient", + # Types "Subscriber", - "UsageStats", + "UsageStats", "PaymentTransaction", "SystemStats", "HealthStatus", "WebSocketMessage", + "UsageEvent", + "RatingPlan", + "RealTimeUsage", + "PaginatedResponse", + # Exceptions "TelecomError", "AuthenticationError", "APIError", "NetworkError", + "ValidationError", + "RateLimitError", + "ServerError", ] diff --git a/sdk/python/telecom_sdk/api.py b/sdk/python/telecom_sdk/api.py new file mode 100644 index 0000000..061b7ee --- /dev/null +++ b/sdk/python/telecom_sdk/api.py @@ -0,0 +1,389 @@ +import asyncio +import logging +from typing import Optional, Dict, Any, List +from datetime import datetime +import aiohttp + +from .auth import AuthProvider +from .exceptions import ( + TelecomError, + AuthenticationError, + APIError, + NetworkError, + RateLimitError, + ServerError, +) +from .types import ( + Subscriber, + UsageStats, + PaymentTransaction, + SystemStats, + HealthStatus, + RealTimeUsage, + PaginatedResponse, + RatingPlan, +) + + +class APIClient: + """HTTP client for making API requests to the Telecom Platform.""" + + def __init__( + self, + base_url: str, + auth_provider: AuthProvider, + timeout: int = 30, + max_retries: int = 3, + retry_delay: float = 1.0, + enable_logging: bool = False, + ): + """ + Initialize the API client. + + Args: + base_url: Base URL for the API server + auth_provider: Authentication provider instance + timeout: Request timeout in seconds + max_retries: Maximum number of retry attempts + retry_delay: Delay between retries in seconds + enable_logging: Enable debug logging + """ + self.base_url = base_url.rstrip('/') + self.auth_provider = auth_provider + self.timeout = aiohttp.ClientTimeout(total=timeout) + self.max_retries = max_retries + self.retry_delay = retry_delay + self.enable_logging = enable_logging + + if enable_logging: + logging.basicConfig(level=logging.DEBUG) + self.logger = logging.getLogger(__name__) + else: + self.logger = logging.getLogger(__name__) + self.logger.disabled = True + + self._session: Optional[aiohttp.ClientSession] = None + + async def __aenter__(self): + """Async context manager entry.""" + await self._ensure_session() + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb): + """Async context manager exit.""" + await self.close() + + async def _ensure_session(self): + """Ensure aiohttp session is created.""" + if self._session is None or self._session.closed: + headers = self.auth_provider.get_headers() + self._session = aiohttp.ClientSession( + headers=headers, + timeout=self.timeout, + ) + + async def make_request( + self, + method: str, + endpoint: str, + data: Optional[Dict[str, Any]] = None, + params: Optional[Dict[str, str]] = None, + ) -> Dict[str, Any]: + """ + Make an HTTP request with retry logic. + + Args: + method: HTTP method + endpoint: API endpoint + data: Request body data + params: Query parameters + + Returns: + Response data as dictionary + + Raises: + TelecomError: On API errors + """ + await self._ensure_session() + url = f"{self.base_url}{endpoint}" + + for attempt in range(self.max_retries + 1): + try: + self.logger.debug(f"Making {method} request to {url} (attempt {attempt + 1})") + + async with self._session.request( + method=method, + url=url, + json=data, + params=params, + ) as response: + await self._handle_response_errors(response) + return await response.json() + + except aiohttp.ClientError as e: + if attempt == self.max_retries: + raise NetworkError(f"Network error after {self.max_retries} retries: {e}") + + self.logger.warning(f"Request failed (attempt {attempt + 1}), retrying in {self.retry_delay}s: {e}") + await asyncio.sleep(self.retry_delay * (2 ** attempt)) + + async def _handle_response_errors(self, response: aiohttp.ClientResponse): + """Handle HTTP response errors.""" + if response.status == 401: + raise AuthenticationError("Authentication failed") + elif response.status == 429: + raise RateLimitError("Rate limit exceeded") + elif 400 <= response.status < 500: + error_data = await response.json().get("error", "Bad request") + raise APIError(f"API error: {error_data}", response.status) + elif response.status >= 500: + raise ServerError(f"Server error: {response.status}", response.status) + + async def close(self) -> None: + """Close the HTTP session.""" + if self._session and not self._session.closed: + await self._session.close() + self.logger.info("API client closed") + + +class SubscriberAPI: + """API endpoints for subscriber management.""" + + def __init__(self, client: APIClient): + self.client = client + + async def get(self, subscriber_id: int) -> Subscriber: + """Get subscriber by ID.""" + response = await self.client.make_request("GET", f"/v1/subscribers/{subscriber_id}") + return Subscriber(**response) + + async def list( + self, + page: int = 1, + page_size: int = 50, + status: Optional[str] = None, + ) -> PaginatedResponse: + """List subscribers with pagination.""" + params = {"page": page, "page_size": page_size} + if status: + params["status"] = status + + response = await self.client.make_request("GET", "/v1/subscribers", params=params) + return PaginatedResponse(**response) + + async def create( + self, + imsi: str, + msisdn: str, + first_name: str, + last_name: str, + email: str, + plan_id: int, + organization_id: Optional[str] = None, + ) -> Subscriber: + """Create a new subscriber.""" + data = { + "imsi": imsi, + "msisdn": msisdn, + "first_name": first_name, + "last_name": last_name, + "email": email, + "plan_id": plan_id, + } + if organization_id: + data["organization_id"] = organization_id + + response = await self.client.make_request("POST", "/v1/subscribers", data=data) + return Subscriber(**response) + + async def update(self, subscriber_id: int, **kwargs) -> Subscriber: + """Update subscriber details.""" + response = await self.client.make_request( + "PUT", + f"/v1/subscribers/{subscriber_id}", + data=kwargs, + ) + return Subscriber(**response) + + async def suspend(self, subscriber_id: int) -> Subscriber: + """Suspend a subscriber.""" + response = await self.client.make_request("POST", f"/v1/subscribers/{subscriber_id}/suspend") + return Subscriber(**response) + + async def activate(self, subscriber_id: int) -> Subscriber: + """Activate a suspended subscriber.""" + response = await self.client.make_request("POST", f"/v1/subscribers/{subscriber_id}/activate") + return Subscriber(**response) + + async def terminate(self, subscriber_id: int) -> bool: + """Terminate a subscriber.""" + response = await self.client.make_request("DELETE", f"/v1/subscribers/{subscriber_id}") + return response.get("success", False) + + +class UsageAPI: + """API endpoints for usage management.""" + + def __init__(self, client: APIClient): + self.client = client + + async def get_stats( + self, + subscriber_id: int, + start_date: datetime, + end_date: datetime, + ) -> UsageStats: + """Get usage statistics for a subscriber.""" + params = { + "start_date": start_date.isoformat(), + "end_date": end_date.isoformat(), + } + response = await self.client.make_request( + "GET", + f"/v1/subscribers/{subscriber_id}/usage", + params=params, + ) + return UsageStats(**response) + + async def list_events( + self, + subscriber_id: Optional[int] = None, + usage_type: Optional[str] = None, + start_date: Optional[datetime] = None, + end_date: Optional[datetime] = None, + page: int = 1, + page_size: int = 50, + ) -> PaginatedResponse: + """List usage events with filtering.""" + params = {"page": page, "page_size": page_size} + + if subscriber_id: + params["subscriber_id"] = subscriber_id + if usage_type: + params["usage_type"] = usage_type + if start_date: + params["start_date"] = start_date.isoformat() + if end_date: + params["end_date"] = end_date.isoformat() + + response = await self.client.make_request("GET", "/v1/usage/events", params=params) + return PaginatedResponse(**response) + + async def get_realtime(self, subscriber_id: int) -> RealTimeUsage: + """Get real-time usage for a subscriber.""" + response = await self.client.make_request("GET", f"/v1/subscribers/{subscriber_id}/realtime") + return RealTimeUsage(**response) + + +class PaymentAPI: + """API endpoints for payment management.""" + + def __init__(self, client: APIClient): + self.client = client + + async def create_transaction( + self, + subscriber_id: int, + amount: float, + currency: str = "USD", + gateway: str = "stripe", + metadata: Optional[Dict[str, Any]] = None, + ) -> PaymentTransaction: + """Create a payment transaction.""" + data = { + "subscriber_id": subscriber_id, + "amount": amount, + "currency": currency, + "gateway": gateway, + } + if metadata: + data["metadata"] = metadata + + response = await self.client.make_request("POST", "/v1/payments/transactions", data=data) + return PaymentTransaction(**response) + + async def get_transaction(self, transaction_id: str) -> PaymentTransaction: + """Get payment transaction by ID.""" + response = await self.client.make_request("GET", f"/v1/payments/transactions/{transaction_id}") + return PaymentTransaction(**response) + + async def list_transactions( + self, + subscriber_id: Optional[int] = None, + status: Optional[str] = None, + page: int = 1, + page_size: int = 50, + ) -> PaginatedResponse: + """List payment transactions with filtering.""" + params = {"page": page, "page_size": page_size} + + if subscriber_id: + params["subscriber_id"] = subscriber_id + if status: + params["status"] = status + + response = await self.client.make_request("GET", "/v1/payments/transactions", params=params) + return PaginatedResponse(**response) + + +class RatingPlanAPI: + """API endpoints for rating plan management.""" + + def __init__(self, client: APIClient): + self.client = client + + async def list(self) -> List[RatingPlan]: + """List all available rating plans.""" + response = await self.client.make_request("GET", "/v1/rating-plans") + return [RatingPlan(**plan) for plan in response] + + async def get(self, plan_id: str) -> RatingPlan: + """Get rating plan by ID.""" + response = await self.client.make_request("GET", f"/v1/rating-plans/{plan_id}") + return RatingPlan(**response) + + +class SystemAPI: + """API endpoints for system management.""" + + def __init__(self, client: APIClient): + self.client = client + + async def get_stats(self) -> SystemStats: + """Get system statistics.""" + response = await self.client.make_request("GET", "/v1/system/stats") + return SystemStats(**response) + + async def get_health(self) -> HealthStatus: + """Get system health status.""" + response = await self.client.make_request("GET", "/v1/health") + return HealthStatus(**response) + + +class GraphQLAPI: + """API endpoints for GraphQL queries.""" + + def __init__(self, client: APIClient): + self.client = client + + async def execute( + self, + query: str, + variables: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: + """ + Execute a GraphQL query. + + Args: + query: GraphQL query string + variables: Optional variables for the query + + Returns: + GraphQL response data + """ + data = {"query": query} + if variables: + data["variables"] = variables + + response = await self.client.make_request("POST", "/graphql", data=data) + return response diff --git a/sdk/python/telecom_sdk/auth.py b/sdk/python/telecom_sdk/auth.py new file mode 100644 index 0000000..d375917 --- /dev/null +++ b/sdk/python/telecom_sdk/auth.py @@ -0,0 +1,120 @@ +from typing import Optional, Dict +import jwt +from datetime import datetime, timedelta + + +class AuthProvider: + """Handles authentication for the Telecom SDK.""" + + def __init__( + self, + api_key: Optional[str] = None, + jwt_secret: Optional[str] = None, + jwt_algorithm: str = "HS256", + ): + """ + Initialize the authentication provider. + + Args: + api_key: API key for authentication + jwt_secret: Secret key for JWT token generation/validation + jwt_algorithm: Algorithm for JWT encoding/decoding + """ + self.api_key = api_key + self.jwt_secret = jwt_secret + self.jwt_algorithm = jwt_algorithm + self._token_cache: Optional[str] = None + self._token_expiry: Optional[datetime] = None + + def get_headers(self) -> Dict[str, str]: + """ + Get authentication headers for API requests. + + Returns: + Dictionary of authentication headers + """ + headers = { + "Content-Type": "application/json", + "User-Agent": "Telecom-Python-SDK/1.0.0", + } + + if self.api_key: + headers["X-API-Key"] = self.api_key + + if self._token_cache and self._is_token_valid(): + headers["Authorization"] = f"Bearer {self._token_cache}" + + return headers + + def generate_jwt_token( + self, + user_id: str, + expiry_hours: int = 24, + **claims + ) -> str: + """ + Generate a JWT token for authentication. + + Args: + user_id: User identifier + expiry_hours: Token expiry time in hours + **claims: Additional claims to include in the token + + Returns: + JWT token string + """ + if not self.jwt_secret: + raise ValueError("JWT secret not configured") + + payload = { + "sub": user_id, + "exp": datetime.utcnow() + timedelta(hours=expiry_hours), + "iat": datetime.utcnow(), + **claims + } + + token = jwt.encode(payload, self.jwt_secret, algorithm=self.jwt_algorithm) + self._token_cache = token + self._token_expiry = datetime.utcnow() + timedelta(hours=expiry_hours) + + return token + + def validate_jwt_token(self, token: str) -> Dict: + """ + Validate a JWT token. + + Args: + token: JWT token string + + Returns: + Decoded token payload + + Raises: + ValueError: If token is invalid or expired + """ + if not self.jwt_secret: + raise ValueError("JWT secret not configured") + + try: + payload = jwt.decode( + token, + self.jwt_secret, + algorithms=[self.jwt_algorithm] + ) + return payload + except jwt.ExpiredSignatureError: + raise ValueError("Token has expired") + except jwt.InvalidTokenError as e: + raise ValueError(f"Invalid token: {e}") + + def _is_token_valid(self) -> bool: + """Check if the cached token is still valid.""" + if not self._token_cache or not self._token_expiry: + return False + + return datetime.utcnow() < self._token_expiry + + def clear_token_cache(self) -> None: + """Clear the cached token.""" + self._token_cache = None + self._token_expiry = None diff --git a/sdk/python/telecom_sdk/client.py b/sdk/python/telecom_sdk/client.py index fad6700..c4601bf 100644 --- a/sdk/python/telecom_sdk/client.py +++ b/sdk/python/telecom_sdk/client.py @@ -1,12 +1,17 @@ -import asyncio -import json import logging -from typing import Optional, Dict, Any, List, Union -from datetime import datetime -import aiohttp -import websockets -from websockets.exceptions import ConnectionClosed +from typing import Optional +from .auth import AuthProvider +from .api import ( + APIClient, + SubscriberAPI, + UsageAPI, + PaymentAPI, + RatingPlanAPI, + SystemAPI, + GraphQLAPI, +) +from .websocket import WebSocketClient from .types import ( Subscriber, UsageStats, @@ -31,11 +36,13 @@ from .exceptions import ( class TelecomSDK: + """Main SDK client for the Telecom Platform.""" def __init__( self, api_url: str = "http://localhost:8000", api_key: Optional[str] = None, + jwt_secret: Optional[str] = None, timeout: int = 30, max_retries: int = 3, retry_delay: float = 1.0, @@ -47,16 +54,13 @@ class TelecomSDK: Args: api_url: Base URL for the API server api_key: API key for authentication + jwt_secret: Secret key for JWT token generation/validation timeout: Request timeout in seconds max_retries: Maximum number of retry attempts retry_delay: Delay between retries in seconds enable_logging: Enable debug logging """ self.api_url = api_url.rstrip('/') - self.api_key = api_key - self.timeout = aiohttp.ClientTimeout(total=timeout) - self.max_retries = max_retries - self.retry_delay = retry_delay self.enable_logging = enable_logging if enable_logging: @@ -66,357 +70,75 @@ class TelecomSDK: self.logger = logging.getLogger(__name__) self.logger.disabled = True - self._session: Optional[aiohttp.ClientSession] = None - self._websocket: Optional[websockets.WebSocketServerProtocol] = None + # Initialize authentication provider + self.auth_provider = AuthProvider( + api_key=api_key, + jwt_secret=jwt_secret, + ) + + # Initialize API client + self.api_client = APIClient( + base_url=api_url, + auth_provider=self.auth_provider, + timeout=timeout, + max_retries=max_retries, + retry_delay=retry_delay, + enable_logging=enable_logging, + ) + + # Initialize WebSocket client + self.websocket_client = WebSocketClient( + api_url=api_url, + auth_provider=self.auth_provider, + enable_logging=enable_logging, + ) + + # Initialize API modules + self.subscribers = SubscriberAPI(self.api_client) + self.usage = UsageAPI(self.api_client) + self.payments = PaymentAPI(self.api_client) + self.rating_plans = RatingPlanAPI(self.api_client) + self.system = SystemAPI(self.api_client) + self.graphql = GraphQLAPI(self.api_client) async def __aenter__(self): """Async context manager entry.""" - await self._ensure_session() + await self.api_client._ensure_session() return self async def __aexit__(self, exc_type, exc_val, exc_tb): """Async context manager exit.""" await self.close() - async def _ensure_session(self): - """Ensure aiohttp session is created.""" - if self._session is None or self._session.closed: - headers = self._get_headers() - self._session = aiohttp.ClientSession( - headers=headers, - timeout=self.timeout, - ) + # Authentication methods - def _get_headers(self) -> Dict[str, str]: - """Get default headers for API requests.""" - headers = { - "Content-Type": "application/json", - "User-Agent": f"Telecom-Python-SDK/1.0.0", - } - if self.api_key: - headers["Authorization"] = f"Bearer {self.api_key}" - return headers - - async def _make_request( + def generate_jwt_token( self, - method: str, - endpoint: str, - data: Optional[Dict[str, Any]] = None, - params: Optional[Dict[str, str]] = None, - ) -> Dict[str, Any]: - """ - Make an HTTP request with retry logic. - - Args: - method: HTTP method - endpoint: API endpoint - data: Request body data - params: Query parameters - - Returns: - Response data as dictionary - - Raises: - TelecomError: On API errors - """ - await self._ensure_session() - url = f"{self.api_url}{endpoint}" - - for attempt in range(self.max_retries + 1): - try: - self.logger.debug(f"Making {method} request to {url} (attempt {attempt + 1})") - - async with self._session.request( - method=method, - url=url, - json=data, - params=params, - ) as response: - await self._handle_response_errors(response) - return await response.json() - - except aiohttp.ClientError as e: - if attempt == self.max_retries: - raise NetworkError(f"Network error after {self.max_retries} retries: {e}") - - self.logger.warning(f"Request failed (attempt {attempt + 1}), retrying in {self.retry_delay}s: {e}") - await asyncio.sleep(self.retry_delay * (2 ** attempt)) + user_id: str, + expiry_hours: int = 24, + **claims + ) -> str: + """Generate a JWT token for authentication.""" + return self.auth_provider.generate_jwt_token(user_id, expiry_hours, **claims) - async def _handle_response_errors(self, response: aiohttp.ClientResponse): - """Handle HTTP response errors.""" - if response.status == 401: - raise AuthenticationError("Authentication failed") - elif response.status == 429: - raise RateLimitError("Rate limit exceeded") - elif 400 <= response.status < 500: - error_data = await response.json().get("error", "Bad request") - raise APIError(f"API error: {error_data}", response.status) - elif response.status >= 500: - raise ServerError(f"Server error: {response.status}", response.status) + def validate_jwt_token(self, token: str) -> dict: + """Validate a JWT token.""" + return self.auth_provider.validate_jwt_token(token) - # Subscriber Management + # WebSocket methods - async def get_subscriber(self, subscriber_id: int) -> Subscriber: - """Get subscriber by ID.""" - response = await self._make_request("GET", f"/v1/subscribers/{subscriber_id}") - return Subscriber(**response) - - async def list_subscribers( - self, - page: int = 1, - page_size: int = 50, - status: Optional[str] = None, - ) -> PaginatedResponse: - """List subscribers with pagination.""" - params = {"page": page, "page_size": page_size} - if status: - params["status"] = status - - response = await self._make_request("GET", "/v1/subscribers", params=params) - return PaginatedResponse(**response) - - async def create_subscriber( - self, - imsi: str, - msisdn: str, - first_name: str, - last_name: str, - email: str, - plan_id: int, - organization_id: Optional[str] = None, - ) -> Subscriber: - """Create a new subscriber.""" - data = { - "imsi": imsi, - "msisdn": msisdn, - "first_name": first_name, - "last_name": last_name, - "email": email, - "plan_id": plan_id, - } - if organization_id: - data["organization_id"] = organization_id - - response = await self._make_request("POST", "/v1/subscribers", data=data) - return Subscriber(**response) - - async def update_subscriber( - self, - subscriber_id: int, - **kwargs, - ) -> Subscriber: - """Update subscriber details.""" - response = await self._make_request( - "PUT", - f"/v1/subscribers/{subscriber_id}", - data=kwargs, - ) - return Subscriber(**response) - - async def suspend_subscriber(self, subscriber_id: int) -> Subscriber: - """Suspend a subscriber.""" - response = await self._make_request("POST", f"/v1/subscribers/{subscriber_id}/suspend") - return Subscriber(**response) - - async def activate_subscriber(self, subscriber_id: int) -> Subscriber: - """Activate a suspended subscriber.""" - response = await self._make_request("POST", f"/v1/subscribers/{subscriber_id}/activate") - return Subscriber(**response) - - async def terminate_subscriber(self, subscriber_id: int) -> bool: - """Terminate a subscriber.""" - response = await self._make_request("DELETE", f"/v1/subscribers/{subscriber_id}") - return response.get("success", False) - - # Usage Management - - async def get_usage_stats( - self, - subscriber_id: int, - start_date: datetime, - end_date: datetime, - ) -> UsageStats: - """Get usage statistics for a subscriber.""" - params = { - "start_date": start_date.isoformat(), - "end_date": end_date.isoformat(), - } - response = await self._make_request( - "GET", - f"/v1/subscribers/{subscriber_id}/usage", - params=params, - ) - return UsageStats(**response) - - async def list_usage_events( - self, - subscriber_id: Optional[int] = None, - usage_type: Optional[str] = None, - start_date: Optional[datetime] = None, - end_date: Optional[datetime] = None, - page: int = 1, - page_size: int = 50, - ) -> PaginatedResponse: - """List usage events with filtering.""" - params = {"page": page, "page_size": page_size} - - if subscriber_id: - params["subscriber_id"] = subscriber_id - if usage_type: - params["usage_type"] = usage_type - if start_date: - params["start_date"] = start_date.isoformat() - if end_date: - params["end_date"] = end_date.isoformat() - - response = await self._make_request("GET", "/v1/usage/events", params=params) - return PaginatedResponse(**response) - - async def get_real_time_usage(self, subscriber_id: int) -> RealTimeUsage: - """Get real-time usage for a subscriber.""" - response = await self._make_request("GET", f"/v1/subscribers/{subscriber_id}/realtime") - return RealTimeUsage(**response) - - # Payment Management - - async def create_payment_transaction( - self, - subscriber_id: int, - amount: float, - currency: str = "USD", - gateway: str = "stripe", - metadata: Optional[Dict[str, Any]] = None, - ) -> PaymentTransaction: - """Create a payment transaction.""" - data = { - "subscriber_id": subscriber_id, - "amount": amount, - "currency": currency, - "gateway": gateway, - } - if metadata: - data["metadata"] = metadata - - response = await self._make_request("POST", "/v1/payments/transactions", data=data) - return PaymentTransaction(**response) - - async def get_payment_transaction(self, transaction_id: str) -> PaymentTransaction: - """Get payment transaction by ID.""" - response = await self._make_request("GET", f"/v1/payments/transactions/{transaction_id}") - return PaymentTransaction(**response) - - async def list_payment_transactions( - self, - subscriber_id: Optional[int] = None, - status: Optional[str] = None, - page: int = 1, - page_size: int = 50, - ) -> PaginatedResponse: - """List payment transactions with filtering.""" - params = {"page": page, "page_size": page_size} - - if subscriber_id: - params["subscriber_id"] = subscriber_id - if status: - params["status"] = status - - response = await self._make_request("GET", "/v1/payments/transactions", params=params) - return PaginatedResponse(**response) - - # Rating Plans - - async def list_rating_plans(self) -> List[RatingPlan]: - """List all available rating plans.""" - response = await self._make_request("GET", "/v1/rating-plans") - return [RatingPlan(**plan) for plan in response] - - async def get_rating_plan(self, plan_id: str) -> RatingPlan: - """Get rating plan by ID.""" - response = await self._make_request("GET", f"/v1/rating-plans/{plan_id}") - return RatingPlan(**response) - - # System Management - - async def get_system_stats(self) -> SystemStats: - """Get system statistics.""" - response = await self._make_request("GET", "/v1/system/stats") - return SystemStats(**response) - - async def get_health_status(self) -> HealthStatus: - """Get system health status.""" - response = await self._make_request("GET", "/v1/health") - return HealthStatus(**response) - - # WebSocket Support - - async def connect_websocket(self, message_handler: callable) -> None: - """ - Connect to WebSocket for real-time updates. - - Args: - message_handler: Async function to handle WebSocket messages - """ - ws_url = self.api_url.replace("http://", "ws://") + "/ws" - - try: - self._websocket = await websockets.connect(ws_url) - self.logger.info("WebSocket connected") - - async for message in self._websocket: - try: - data = json.loads(message) - ws_message = WebSocketMessage(**data) - await message_handler(ws_message) - except json.JSONDecodeError as e: - self.logger.error(f"Failed to decode WebSocket message: {e}") - except Exception as e: - self.logger.error(f"Error handling WebSocket message: {e}") - - except ConnectionClosed: - self.logger.warning("WebSocket connection closed") - except Exception as e: - self.logger.error(f"WebSocket error: {e}") - raise NetworkError(f"WebSocket connection failed: {e}") + async def connect_websocket(self, message_handler) -> None: + """Connect to WebSocket for real-time updates.""" + await self.websocket_client.connect(message_handler) async def disconnect_websocket(self) -> None: """Disconnect from WebSocket.""" - if self._websocket: - await self._websocket.close() - self._websocket = None - self.logger.info("WebSocket disconnected") + await self.websocket_client.disconnect() - # GraphQL Support - - async def execute_graphql_query( - self, - query: str, - variables: Optional[Dict[str, Any]] = None, - ) -> Dict[str, Any]: - """ - Execute a GraphQL query. - - Args: - query: GraphQL query string - variables: Optional variables for the query - - Returns: - GraphQL response data - """ - data = {"query": query} - if variables: - data["variables"] = variables - - response = await self._make_request("POST", "/graphql", data=data) - return response - - # Utility Methods + # Utility methods async def close(self) -> None: """Close the SDK client and cleanup resources.""" - if self._websocket: - await self.disconnect_websocket() - - if self._session and not self._session.closed: - await self._session.close() - + await self.websocket_client.disconnect() + await self.api_client.close() self.logger.info("Telecom SDK client closed") diff --git a/sdk/python/telecom_sdk/websocket.py b/sdk/python/telecom_sdk/websocket.py new file mode 100644 index 0000000..5220587 --- /dev/null +++ b/sdk/python/telecom_sdk/websocket.py @@ -0,0 +1,100 @@ +import json +import logging +from typing import Optional, Callable +import websockets +from websockets.exceptions import ConnectionClosed + +from .types import WebSocketMessage +from .exceptions import NetworkError + + +class WebSocketClient: + """WebSocket client for real-time updates from the Telecom Platform.""" + + def __init__( + self, + api_url: str, + auth_provider, + enable_logging: bool = False, + ): + """ + Initialize the WebSocket client. + + Args: + api_url: Base URL for the API server + auth_provider: Authentication provider instance + enable_logging: Enable debug logging + """ + self.api_url = api_url.rstrip('/') + self.auth_provider = auth_provider + self.enable_logging = enable_logging + + if enable_logging: + self.logger = logging.getLogger(__name__) + else: + self.logger = logging.getLogger(__name__) + self.logger.disabled = True + + self._websocket: Optional[websockets.WebSocketServerProtocol] = None + self._connected = False + + async def connect(self, message_handler: Callable[[WebSocketMessage], None]) -> None: + """ + Connect to WebSocket for real-time updates. + + Args: + message_handler: Async function to handle WebSocket messages + """ + ws_url = self.api_url.replace("http://", "ws://") + "/ws" + headers = self.auth_provider.get_headers() + + try: + self._websocket = await websockets.connect(ws_url, extra_headers=headers) + self._connected = True + self.logger.info("WebSocket connected") + + async for message in self._websocket: + try: + data = json.loads(message) + ws_message = WebSocketMessage(**data) + await message_handler(ws_message) + except json.JSONDecodeError as e: + self.logger.error(f"Failed to decode WebSocket message: {e}") + except Exception as e: + self.logger.error(f"Error handling WebSocket message: {e}") + + except ConnectionClosed: + self.logger.warning("WebSocket connection closed") + self._connected = False + except Exception as e: + self.logger.error(f"WebSocket error: {e}") + self._connected = False + raise NetworkError(f"WebSocket connection failed: {e}") + + async def disconnect(self) -> None: + """Disconnect from WebSocket.""" + if self._websocket: + await self._websocket.close() + self._websocket = None + self._connected = False + self.logger.info("WebSocket disconnected") + + def is_connected(self) -> bool: + """Check if WebSocket is connected.""" + return self._connected + + async def send(self, data: dict) -> None: + """ + Send data through the WebSocket connection. + + Args: + data: Dictionary data to send + """ + if not self._connected or not self._websocket: + raise NetworkError("WebSocket is not connected") + + try: + await self._websocket.send(json.dumps(data)) + except Exception as e: + self.logger.error(f"Failed to send WebSocket message: {e}") + raise NetworkError(f"Failed to send message: {e}")