feat: Add Python SDK with async/await support and comprehensive Telecom Platform API integration

- Add TelecomSDK class with async context manager and aiohttp ClientSession with 30s timeout, 3 retries
- Add subscriber management: get_subscriber, list_subscribers with pagination/status filter, create_subscriber, update_subscriber, suspend_subscriber, activate_subscriber, terminate_subscriber
- Add usage tracking: get_usage_stats with date range, list_usage_events with filtering, get_real_time_usage
This commit is contained in:
nutcas3
2026-04-17 17:49:05 +03:00
parent 63506fdc24
commit 7aab9a058b
6 changed files with 930 additions and 0 deletions
+300
View File
@@ -0,0 +1,300 @@
# Telecom Platform Python SDK
Async Python SDK for the Telecom Platform with full type hints and modern async/await support.
## Features
- **Async/Await Support**: Full async/await support with aiohttp
- **Type Safety**: Complete type hints with Pydantic models
- **WebSocket Support**: Real-time updates via WebSocket
- **GraphQL Support**: Execute GraphQL queries
- **Retry Logic**: Built-in retry mechanism with exponential backoff
- **Error Handling**: Comprehensive error handling with custom exceptions
- **Logging**: Optional debug logging
- **Context Manager**: Async context manager support
## Installation
```bash
pip install telecom-sdk
```
Or install from source:
```bash
git clone https://github.com/nutcas3/telecom-platform.git
cd telecom-platform/sdk/python
# Use uv for faster installation (recommended)
uv pip install -r requirements.txt
uv pip install -e .
# Or traditional pip
pip install -r requirements.txt
pip install -e .
```
## Quick Start
```python
import asyncio
from telecom_sdk import TelecomSDK
async def main():
async with TelecomSDK(
api_url="http://localhost:8000",
api_key="your-api-key",
enable_logging=True
) as sdk:
# Get subscriber
subscriber = await sdk.get_subscriber(1)
print(f"Subscriber: {subscriber.first_name} {subscriber.last_name}")
# Get usage stats
from datetime import datetime, timedelta
end_date = datetime.now()
start_date = end_date - timedelta(days=30)
usage = await sdk.get_usage_stats(1, start_date, end_date)
print(f"Data usage: {usage.data_up + usage.data_down} bytes")
# Create payment transaction
transaction = await sdk.create_payment_transaction(
subscriber_id=1,
amount=25.00,
currency="USD"
)
print(f"Transaction ID: {transaction.id}")
if __name__ == "__main__":
asyncio.run(main())
```
## WebSocket Example
```python
import asyncio
from telecom_sdk import TelecomSDK, WebSocketMessage
async def handle_message(message: WebSocketMessage):
"""Handle WebSocket messages."""
print(f"Received message type: {message.type}")
print(f"Message data: {message.data}")
async def main():
sdk = TelecomSDK(api_url="http://localhost:8000")
# Connect to WebSocket
await sdk.connect_websocket(handle_message)
# Keep running
try:
while True:
await asyncio.sleep(1)
except KeyboardInterrupt:
await sdk.disconnect_websocket()
if __name__ == "__main__":
asyncio.run(main())
```
## GraphQL Example
```python
import asyncio
from telecom_sdk import TelecomSDK
async def main():
async with TelecomSDK(api_url="http://localhost:8000") as sdk:
query = """
query GetSubscribers($first: Int, $after: String) {
subscribers(first: $first, after: $after) {
edges {
node {
id
firstName
lastName
email
status
}
}
pageInfo {
hasNextPage
endCursor
}
}
}
"""
variables = {"first": 10}
result = await sdk.execute_graphql_query(query, variables)
for edge in result["data"]["subscribers"]["edges"]:
subscriber = edge["node"]
print(f"{subscriber['firstName']} {subscriber['lastName']}")
if __name__ == "__main__":
asyncio.run(main())
```
## API Reference
### Subscriber Management
```python
# Get subscriber
subscriber = await sdk.get_subscriber(subscriber_id)
# List subscribers
subscribers = await sdk.list_subscribers(page=1, page_size=50, status="active")
# Create subscriber
subscriber = await sdk.create_subscriber(
imsi="208930000000001",
msisdn="+1234567890",
first_name="John",
last_name="Doe",
email="[email protected]",
plan_id=1
)
# Update subscriber
subscriber = await sdk.update_subscriber(subscriber_id, first_name="Jane")
# Suspend subscriber
subscriber = await sdk.suspend_subscriber(subscriber_id)
# Activate subscriber
subscriber = await sdk.activate_subscriber(subscriber_id)
# Terminate subscriber
success = await sdk.terminate_subscriber(subscriber_id)
```
### Usage Management
```python
from datetime import datetime, timedelta
# Get usage stats
usage = await sdk.get_usage_stats(
subscriber_id=1,
start_date=datetime.now() - timedelta(days=30),
end_date=datetime.now()
)
# List usage events
events = await sdk.list_usage_events(
subscriber_id=1,
usage_type="data",
page=1,
page_size=50
)
# Get real-time usage
realtime = await sdk.get_real_time_usage(subscriber_id)
```
### Payment Management
```python
# Create payment transaction
transaction = await sdk.create_payment_transaction(
subscriber_id=1,
amount=25.00,
currency="USD",
gateway="stripe",
metadata={"description": "Monthly plan"}
)
# Get transaction
transaction = await sdk.get_payment_transaction(transaction_id)
# List transactions
transactions = await sdk.list_payment_transactions(
subscriber_id=1,
status="completed"
)
```
### System Management
```python
# Get system stats
stats = await sdk.get_system_stats()
# Get health status
health = await sdk.get_health_status()
```
### Rating Plans
```python
# List rating plans
plans = await sdk.list_rating_plans()
# Get specific plan
plan = await sdk.get_rating_plan(plan_id)
```
## Error Handling
```python
from telecom_sdk import TelecomSDK, AuthenticationError, APIError, NetworkError
async def main():
try:
async with TelecomSDK(api_url="http://localhost:8000") as sdk:
subscriber = await sdk.get_subscriber(1)
except AuthenticationError:
print("Authentication failed")
except APIError as e:
print(f"API error: {e}")
except NetworkError as e:
print(f"Network error: {e}")
except TelecomError as e:
print(f"Telecom error: {e}")
```
## Configuration
The SDK can be configured with the following options:
- `api_url`: Base URL for the API server (default: "http://localhost:8000")
- `api_key`: API key for authentication
- `timeout`: Request timeout in seconds (default: 30)
- `max_retries`: Maximum number of retry attempts (default: 3)
- `retry_delay`: Delay between retries in seconds (default: 1.0)
- `enable_logging`: Enable debug logging (default: False)
## Development
### Running Tests
```bash
# Install development dependencies
pip install -r requirements-dev.txt
# Run tests
pytest tests/
# Run tests with coverage
pytest tests/ --cov=telecom_sdk
```
### Code Style
```bash
# Format code
black telecom_sdk/
# Lint code
flake8 telecom_sdk/
# Type checking
mypy telecom_sdk/
```
## License
This SDK is licensed under the MIT License.
+6
View File
@@ -0,0 +1,6 @@
# Install with: uv pip install -r requirements.txt
# Or traditional: pip install -r requirements.txt
aiohttp>=3.8.0
websockets>=11.0.0
pydantic>=2.0.0
typing-extensions>=4.0.0
+30
View File
@@ -0,0 +1,30 @@
from .client import TelecomSDK
from .types import (
Subscriber,
UsageStats,
PaymentTransaction,
SystemStats,
HealthStatus,
WebSocketMessage,
)
from .exceptions import (
TelecomError,
AuthenticationError,
APIError,
NetworkError,
)
__version__ = "1.0.0"
__all__ = [
"TelecomSDK",
"Subscriber",
"UsageStats",
"PaymentTransaction",
"SystemStats",
"HealthStatus",
"WebSocketMessage",
"TelecomError",
"AuthenticationError",
"APIError",
"NetworkError",
]
+422
View File
@@ -0,0 +1,422 @@
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 .types import (
Subscriber,
UsageStats,
PaymentTransaction,
SystemStats,
HealthStatus,
WebSocketMessage,
UsageEvent,
RatingPlan,
RealTimeUsage,
PaginatedResponse,
)
from .exceptions import (
TelecomError,
AuthenticationError,
APIError,
NetworkError,
ValidationError,
RateLimitError,
ServerError,
)
class TelecomSDK:
def __init__(
self,
api_url: str = "http://localhost:8000",
api_key: Optional[str] = None,
timeout: int = 30,
max_retries: int = 3,
retry_delay: float = 1.0,
enable_logging: bool = False,
):
"""
Initialize the Telecom SDK client.
Args:
api_url: Base URL for the API server
api_key: API key for authentication
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:
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
self._websocket: Optional[websockets.WebSocketServerProtocol] = 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._get_headers()
self._session = aiohttp.ClientSession(
headers=headers,
timeout=self.timeout,
)
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(
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))
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)
# Subscriber Management
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 disconnect_websocket(self) -> None:
"""Disconnect from WebSocket."""
if self._websocket:
await self._websocket.close()
self._websocket = None
self.logger.info("WebSocket disconnected")
# 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
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()
self.logger.info("Telecom SDK client closed")
+31
View File
@@ -0,0 +1,31 @@
class TelecomError(Exception):
"""Base exception for all Telecom SDK errors."""
def __init__(self, message: str, status_code: Optional[int] = None, response_data: Optional[dict] = None):
super().__init__(message)
self.status_code = status_code
self.response_data = response_data or {}
class AuthenticationError(TelecomError):
pass
class APIError(TelecomError):
pass
class NetworkError(TelecomError):
pass
class ValidationError(TelecomError):
pass
class RateLimitError(TelecomError):
pass
class ServerError(TelecomError):
pass
+141
View File
@@ -0,0 +1,141 @@
from typing import Optional, Dict, Any, List
from datetime import datetime
from pydantic import BaseModel, Field
from enum import Enum
class SubscriberStatus(str, Enum):
ACTIVE = "active"
SUSPENDED = "suspended"
TERMINATED = "terminated"
class UsageType(str, Enum):
DATA = "data"
VOICE = "voice"
SMS = "sms"
class PaymentStatus(str, Enum):
PENDING = "pending"
COMPLETED = "completed"
FAILED = "failed"
REFUNDED = "refunded"
class Subscriber(BaseModel):
id: int
imsi: str
msisdn: str
first_name: str
last_name: str
email: str
organization_id: Optional[str] = None
status: SubscriberStatus
plan_id: int
balance: float = Field(ge=0.0)
created_at: datetime
updated_at: datetime
class Config:
use_enum_values = True
class UsageStats(BaseModel):
subscriber_id: str
data_up: int = Field(ge=0)
data_down: int = Field(ge=0)
voice_seconds: int = Field(ge=0)
sms_count: int = Field(ge=0)
period_start: datetime
period_end: datetime
cost: float = Field(ge=0.0)
class PaymentTransaction(BaseModel):
id: str
subscriber_id: str
amount: float = Field(ge=0.0)
currency: str = Field(min_length=3, max_length=3)
status: PaymentStatus
gateway: str
transaction_id: Optional[str] = None
created_at: datetime
completed_at: Optional[datetime] = None
metadata: Optional[Dict[str, Any]] = None
class SystemStats(BaseModel):
active_sessions: int = Field(ge=0)
total_accounts: int = Field(ge=0)
blocked_users: int = Field(ge=0)
low_balance_alerts: int = Field(ge=0)
uptime: float = Field(ge=0.0)
cpu_usage: float = Field(ge=0.0, le=100.0)
memory_usage: float = Field(ge=0.0, le=100.0)
class HealthStatus(BaseModel):
status: str # "healthy", "degraded", "unhealthy"
timestamp: datetime
checks: Dict[str, Dict[str, Any]]
uptime: float = Field(ge=0.0)
class WebSocketMessage(BaseModel):
type: str
data: Dict[str, Any]
timestamp: datetime
class UsageEvent(BaseModel):
id: str
subscriber_id: str
usage_type: UsageType
amount: int = Field(ge=0)
cost: float = Field(ge=0.0)
timestamp: datetime
metadata: Optional[Dict[str, Any]] = None
class RatingPlan(BaseModel):
plan_id: str
name: str
data_rate: float = Field(ge=0.0)
voice_rate: float = Field(ge=0.0)
sms_rate: float = Field(ge=0.0)
monthly_fee: float = Field(ge=0.0)
data_limit: int = Field(ge=0)
voice_limit: int = Field(ge=0)
sms_limit: int = Field(ge=0)
class CurrentSession(BaseModel):
session_id: str
start_time: datetime
data_up: int = Field(ge=0)
data_down: int = Field(ge=0)
voice_seconds: int = Field(ge=0)
sms_count: int = Field(ge=0)
class RealTimeUsage(BaseModel):
"""Real-time usage model."""
current_session: Optional[CurrentSession] = None
today_usage: Optional[Dict[str, int]] = None
class APIResponse(BaseModel):
"""Generic API response model."""
success: bool
data: Optional[Any] = None
error: Optional[str] = None
message: Optional[str] = None
class PaginatedResponse(BaseModel):
"""Paginated response model."""
items: List[Any]
total: int = Field(ge=0)
page: int = Field(ge=1)
page_size: int = Field(ge=1, le=100)
has_next: bool
has_prev: bool