""" eSIM Atlas Partner API SDK (Python) A lightweight, zero-dependency Python SDK for the eSIM Atlas Partner API. Requires Python 3.8+ and only uses the standard library (urllib, json, hmac). Handles OAuth 2.0 client-credentials authentication automatically: tokens are cached in memory and refreshed transparently 60 seconds before expiry. version: 1.0.0 license: Proprietary see: https://docs.gstarlink.com Usage: from esim_atlas_sdk import EsimAtlasClient client = EsimAtlasClient( client_id="sk_live_abc123", client_secret="your_secret_here", ) products = client.list_products(region="Asia") for product in products["data"]: print(product["name"]) """ from __future__ import annotations import hashlib import hmac import json import time import urllib.error import urllib.parse import urllib.request from typing import Any, Dict, Optional DEFAULT_BASE_URL = "https://api.gstarlink.com" DEFAULT_SCOPE = ( "read:products write:orders read:orders read:esims " "write:webhooks read:webhooks" ) _TOKEN_EXPIRY_BUFFER = 60 # refresh 60s before expiry class ApiError(Exception): """Raised for API errors (RFC 7807 Problem Detail responses).""" def __init__(self, problem: Dict[str, Any]): self.status: int = int(problem.get("status") or 0) self.type: str = str(problem.get("type") or "") self.title: str = str(problem.get("title") or "") self.detail: str = str(problem.get("detail") or "") self.instance: Optional[str] = problem.get("instance") self.errors: Optional[Dict[str, Any]] = problem.get("errors") super().__init__(self.detail or self.title) class OAuthError(Exception): """Raised for OAuth token-endpoint failures.""" def __init__(self, status: int, code: str, description: str): self.status = status self.code = code super().__init__(description or code) class EsimAtlasClient: """eSIM Atlas Partner API client.""" def __init__( self, client_id: str, client_secret: str, base_url: Optional[str] = None, scope: Optional[str] = None, ): """ Args: client_id: OAuth client_id (API key). client_secret: OAuth client_secret. base_url: API base URL (defaults to https://api.gstarlink.com). scope: OAuth scopes (defaults to all scopes). """ self._client_id = client_id self._client_secret = client_secret self._base_url = (base_url or DEFAULT_BASE_URL).rstrip("/") self._scope = scope or DEFAULT_SCOPE self._access_token: Optional[str] = None self._token_expires_at: float = 0.0 # ------------------------------------------------------------------------- # Products # ------------------------------------------------------------------------- def list_products(self, **params: Any) -> Dict[str, Any]: """List products (region, type, status, sort_by, sort_order, limit, cursor).""" return self._request("GET", "/api/v1/products", query=params) def get_product(self, product_id: str) -> Dict[str, Any]: """Get detailed information about a specific product.""" return self._request("GET", f"/api/v1/products/{_encode(product_id)}") # ------------------------------------------------------------------------- # Orders # ------------------------------------------------------------------------- def create_order(self, data: Dict[str, Any], idempotency_key: str) -> Dict[str, Any]: """Create an order. The idempotency key prevents duplicates — reusing the same key returns the original order.""" return self._request( "POST", "/api/v1/orders", body=data, headers={"Idempotency-Key": idempotency_key}, ) def list_orders(self, **params: Any) -> Dict[str, Any]: """List your orders (status, limit, cursor).""" return self._request("GET", "/api/v1/orders", query=params) def get_order(self, order_id: str) -> Dict[str, Any]: """Get a single order with full fulfillment details.""" return self._request("GET", f"/api/v1/orders/{_encode(order_id)}") def get_order_esims(self, order_id: str) -> Dict[str, Any]: """Get eSIM delivery details (ICCID, LPA code, QR URL) for a fulfilled order.""" return self._request("GET", f"/api/v1/orders/{_encode(order_id)}/esims") def get_activation_status(self, order_id: str) -> Dict[str, Any]: """Get activation status for all scheduled activations on an order.""" return self._request("GET", f"/api/v1/orders/{_encode(order_id)}/activation-status") def create_bulk_orders(self, data: Dict[str, Any]) -> Dict[str, Any]: """Create up to 100 orders in one request. Each order is processed independently; individual failures do not affect the rest of the batch.""" return self._request("POST", "/api/v1/orders/bulk", body=data) # ------------------------------------------------------------------------- # Activations (Physical SIM) # ------------------------------------------------------------------------- def submit_activation(self, data: Dict[str, Any]) -> Dict[str, Any]: """Submit a physical SIM activation request with KYC data. The carrier is auto-detected from the ICCID prefix.""" return self._request("POST", "/api/v1/activations", body=data) def list_activations(self, **params: Any) -> Dict[str, Any]: """List your physical SIM activation requests (iccid, status, limit).""" return self._request("GET", "/api/v1/activations", query=params) def get_activation(self, activation_id: str) -> Dict[str, Any]: """Get detailed status of a specific activation request.""" return self._request("GET", f"/api/v1/activations/{_encode(activation_id)}") # ------------------------------------------------------------------------- # eSIMs # ------------------------------------------------------------------------- def get_esim_status(self, iccid: str) -> Dict[str, Any]: """Check activation status, data usage, and current location of an eSIM.""" return self._request("GET", f"/api/v1/esims/{_encode(iccid)}/status") # ------------------------------------------------------------------------- # Inventory # ------------------------------------------------------------------------- def get_balance(self) -> Dict[str, Any]: """Current pre-funded wallet balance per currency. Production orders are charged to this wallet and fail with 402 if it is underfunded.""" return self._request("GET", "/api/v1/balance") def get_inventory_status(self, **params: Any) -> Dict[str, Any]: """Check real-time stock and availability (product_ids, skus, region).""" return self._request("GET", "/api/v1/inventory/status", query=params) # ------------------------------------------------------------------------- # Webhooks # ------------------------------------------------------------------------- def create_webhook(self, data: Dict[str, Any]) -> Dict[str, Any]: """Register a webhook endpoint. The signing secret is returned only once — store it securely.""" return self._request("POST", "/api/v1/webhooks", body=data) def list_webhooks(self, **params: Any) -> Dict[str, Any]: """List your registered webhooks (status, environment, limit, cursor).""" return self._request("GET", "/api/v1/webhooks", query=params) def get_webhook(self, webhook_id: str) -> Dict[str, Any]: """Get detailed information about a specific webhook.""" return self._request("GET", f"/api/v1/webhooks/{_encode(webhook_id)}") def delete_webhook(self, webhook_id: str) -> None: """Delete a webhook registration. This action is irreversible.""" self._request("DELETE", f"/api/v1/webhooks/{_encode(webhook_id)}") # ------------------------------------------------------------------------- # Catalog # ------------------------------------------------------------------------- def sync_catalog(self, **params: Any) -> Dict[str, Any]: """Export the product catalog for e-commerce platforms (format, markup, region, type, include_images).""" return self._request("GET", "/api/v1/catalog/sync", query=params) # ------------------------------------------------------------------------- # Internal # ------------------------------------------------------------------------- def _token(self) -> str: """Obtain or refresh the OAuth access token (cached until 60s before expiry).""" if self._access_token and time.time() < self._token_expires_at: return self._access_token status, body = self._raw_request( "POST", f"{self._base_url}/api/v1/oauth/token", {"Content-Type": "application/json", "Accept": "application/json"}, json.dumps( { "grant_type": "client_credentials", "client_id": self._client_id, "client_secret": self._client_secret, "scope": self._scope, } ).encode("utf-8"), ) if not 200 <= status < 300: code = body.get("error") if isinstance(body, dict) else None desc = body.get("error_description") if isinstance(body, dict) else None raise OAuthError(status, code or "server_error", desc or f"HTTP {status}") self._access_token = body["access_token"] self._token_expires_at = time.time() + int(body["expires_in"]) - _TOKEN_EXPIRY_BUFFER return self._access_token def _request( self, method: str, path: str, query: Optional[Dict[str, Any]] = None, body: Optional[Any] = None, headers: Optional[Dict[str, str]] = None, ) -> Any: url = f"{self._base_url}{path}" if query: filtered = {k: v for k, v in query.items() if v is not None} if filtered: url += "?" + urllib.parse.urlencode(filtered) req_headers = { "Authorization": f"Bearer {self._token()}", "Accept": "application/json", } if headers: req_headers.update(headers) payload = None if body is not None: req_headers["Content-Type"] = "application/json" payload = json.dumps(body).encode("utf-8") status, decoded = self._raw_request(method, url, req_headers, payload) if status == 204: return None if not 200 <= status < 300: if isinstance(decoded, dict) and "type" in decoded and "status" in decoded: raise ApiError(decoded) raise ApiError( { "type": "https://docs.gstarlink.com/errors/unknown", "title": f"HTTP {status}", "status": status, "detail": "" if decoded is None else json.dumps(decoded), } ) return decoded @staticmethod def _raw_request(method: str, url: str, headers: Dict[str, str], body: Optional[bytes]): """Perform a raw HTTP request. Returns (status_code, decoded_json_or_None).""" req = urllib.request.Request(url, data=body, method=method) for key, value in headers.items(): req.add_header(key, value) try: with urllib.request.urlopen(req, timeout=30) as res: raw = res.read().decode("utf-8") status = res.status except urllib.error.HTTPError as err: raw = err.read().decode("utf-8") status = err.code decoded = json.loads(raw) if raw else None return status, decoded def verify_webhook_signature(payload: str, signature: str, secret: str) -> bool: """Verify the authenticity of an incoming webhook request. The signature header format is: ``t=,v1=``. Computes HMAC-SHA256 of ``.`` with the signing secret and compares it against the provided signature in constant time. Args: payload: Raw request body as a string. signature: Value of the ``X-Webhook-Signature`` header. secret: Webhook signing secret (from create_webhook). """ parts: Dict[str, str] = {} for segment in signature.split(","): idx = segment.find("=") if idx > 0: parts[segment[:idx]] = segment[idx + 1:] timestamp = parts.get("t") signature_hex = parts.get("v1") if not timestamp or not signature_hex: return False expected = hmac.new( secret.encode("utf-8"), f"{timestamp}.{payload}".encode("utf-8"), hashlib.sha256, ).hexdigest() return hmac.compare_digest(expected, signature_hex) def _encode(value: str) -> str: return urllib.parse.quote(str(value), safe="")