tessl/pypi-twilio
Twilio API client and TwiML generator for comprehensive telecommunications services
infrastructure.mddocs/
Infrastructure
Low-level infrastructure including HTTP clients, authentication strategies, credential providers, base classes, and exception handling. These components support the high-level APIs and can be customized for advanced use cases.
Capabilities
HTTP Client Infrastructure
Customizable HTTP client system supporting synchronous and asynchronous requests.
class HttpClient:
"""Abstract HTTP client base class"""
def request(
self,
method: str,
uri: str,
data: dict = None,
headers: dict = None,
auth: tuple = None,
timeout: float = None,
allow_redirects: bool = None
) -> 'Response':
"""
Make HTTP request.
Args:
method (str): HTTP method ('GET', 'POST', etc.)
uri (str): Request URI
data (dict): Request body data
headers (dict): HTTP headers
auth (tuple): Authentication tuple (username, password)
timeout (float): Request timeout in seconds
allow_redirects (bool): Follow redirects
Returns:
Response: HTTP response object
"""
class TwilioHttpClient(HttpClient):
"""Default HTTP client using requests library"""
def __init__(
self,
pool_connections: int = 1,
pool_maxsize: int = 1,
max_retries: int = 3,
backoff_factor: float = 0.3,
proxy: dict = None,
ssl_context: 'SSLContext' = None,
timeout: float = 60.0
):
"""
Initialize HTTP client.
Args:
pool_connections (int): Number of connection pools
pool_maxsize (int): Max connections per pool
max_retries (int): Retry attempts for failed requests
backoff_factor (float): Backoff multiplier for retries
proxy (dict): Proxy configuration
ssl_context (SSLContext): Custom SSL context
timeout (float): Default timeout in seconds
"""
class AsyncHttpClient:
"""Abstract asynchronous HTTP client"""
async def request(
self,
method: str,
uri: str,
data: dict = None,
headers: dict = None,
auth: tuple = None,
timeout: float = None
) -> 'Response':
"""Make async HTTP request"""
class TwilioAsyncHttpClient(AsyncHttpClient):
"""Default async HTTP client using aiohttp"""
def __init__(
self,
timeout: float = 60.0,
pool_connections: int = 100,
pool_maxsize: int = 100,
ssl_context: 'SSLContext' = None
):
"""
Initialize async HTTP client.
Args:
timeout (float): Default timeout in seconds
pool_connections (int): Connection pool size
pool_maxsize (int): Maximum pool size
ssl_context (SSLContext): Custom SSL context
"""
async def request(
self,
method: str,
uri: str,
data: dict = None,
headers: dict = None,
auth: tuple = None,
timeout: float = None
) -> 'Response':
"""
Make async HTTP request.
Args:
method (str): HTTP method
uri (str): Request URI
data (dict): Request data
headers (dict): Request headers
auth (tuple): Authentication credentials
timeout (float): Request timeout
Returns:
Response: HTTP response object
"""
class Request:
"""HTTP request representation"""
def __init__(
self,
method: str,
uri: str,
data: dict = None,
headers: dict = None,
params: dict = None,
auth: tuple = None
):
"""
Args:
method (str): HTTP method
uri (str): Request URI
data (dict): Request body data
headers (dict): HTTP headers
params (dict): Query parameters
auth (tuple): Authentication credentials
"""
self.method = method
self.uri = uri
self.data = data
self.headers = headers
self.params = params
self.auth = auth
class Response:
"""HTTP response representation"""
def __init__(
self,
status_code: int,
text: str,
headers: dict = None
):
"""
Args:
status_code (int): HTTP status code
text (str): Response body text
headers (dict): Response headers
"""
self.status_code = status_code
self.text = text
self.headers = headers or {}
@property
def ok(self) -> bool:
"""True if status code indicates success (200-299)"""
return 200 <= self.status_code < 300
@property
def content(self) -> bytes:
"""Response body as bytes"""
return self.text.encode('utf-8')Authentication Strategies
Pluggable authentication system supporting multiple auth methods.
class AuthStrategy:
"""Base authentication strategy"""
def get_auth_string(self) -> str:
"""
Generate authentication string.
Returns:
str: Authentication header value
"""
raise NotImplementedError
class TokenAuthStrategy(AuthStrategy):
"""Token-based authentication (Basic Auth with Account SID/Auth Token)"""
def __init__(self, username: str, password: str):
"""
Args:
username (str): Account SID or API Key SID
password (str): Auth Token or API Key Secret
"""
self.username = username
self.password = password
def get_auth_string(self) -> str:
"""Generate Basic Auth header value"""
class NoAuthStrategy(AuthStrategy):
"""No authentication strategy"""
def get_auth_string(self) -> str:
"""Return empty auth string"""
return ''
class AuthType:
"""Authentication type enumeration"""
BASIC = 'basic'
TOKEN = 'token'
OAUTH = 'oauth'Credential Providers
Alternative authentication mechanisms for advanced scenarios.
class CredentialProvider:
"""Base credential provider"""
def get_headers(self) -> dict:
"""
Get authentication headers.
Returns:
dict: HTTP headers for authentication
"""
raise NotImplementedError
def get_username(self) -> str:
"""Get username for basic auth"""
raise NotImplementedError
def get_password(self) -> str:
"""Get password for basic auth"""
raise NotImplementedError
class ClientCredentialProvider(CredentialProvider):
"""OAuth 2.0 Client Credentials flow provider"""
def __init__(
self,
client_id: str,
client_secret: str,
token_endpoint: str,
scopes: list = None
):
"""
Args:
client_id (str): OAuth client ID
client_secret (str): OAuth client secret
token_endpoint (str): Token endpoint URL
scopes (list): Requested scopes
"""
class OrgsCredentialProvider(CredentialProvider):
"""Organization-based credential provider"""
def __init__(
self,
client_id: str,
client_secret: str,
account_sid: str,
orgs_token_endpoint: str = None
):
"""
Args:
client_id (str): Client ID
client_secret (str): Client secret
account_sid (str): Target account SID
orgs_token_endpoint (str): Organizations token endpoint
"""Base Resource Classes
Foundational classes for REST API resource management.
class ClientBase:
"""Base client implementation"""
def __init__(
self,
username: str = None,
password: str = None,
account_sid: str = None,
region: str = None,
http_client: HttpClient = None,
environment: dict = None,
edge: str = None
): ...
class Domain:
"""Service domain base class"""
def __init__(self, twilio: ClientBase):
self.twilio = twilio
self.base_url = None
def request(
self,
method: str,
uri: str,
data: dict = None,
headers: dict = None
) -> Response:
"""Make authenticated request to domain"""
class Version:
"""API version handling"""
def __init__(self, domain: Domain, version: str = None):
self.domain = domain
self.version = version
def request(
self,
method: str,
uri: str,
data: dict = None,
headers: dict = None
) -> Response:
"""Make versioned API request"""
class InstanceResource:
"""Individual resource instance"""
def __init__(self, version: Version, payload: dict, **kwargs):
self._version = version
self._properties = payload
def fetch(self) -> 'InstanceResource':
"""Fetch latest resource data"""
def delete(self) -> bool:
"""Delete the resource"""
def update(self, **kwargs) -> 'InstanceResource':
"""Update the resource"""
class ListResource:
"""Resource collection"""
def __init__(self, version: Version):
self._version = version
def create(self, **kwargs) -> InstanceResource:
"""Create new resource instance"""
def list(self, limit: int = None, page_size: int = None) -> Iterator[InstanceResource]:
"""List resource instances"""
def stream(self, limit: int = None, page_size: int = None) -> Iterator[InstanceResource]:
"""Stream resource instances"""
class InstanceContext:
"""Resource context for operations"""
def __init__(self, version: Version, payload: dict, **kwargs):
self._version = version
self._properties = payload
def fetch(self) -> InstanceResource:
"""Fetch resource instance"""
def delete(self) -> bool:
"""Delete resource"""
def update(self, **kwargs) -> InstanceResource:
"""Update resource"""
class Page:
"""Paginated results container"""
def __init__(self, version: Version, response: Response):
self._version = version
self._response = response
def get_instance(self, payload: dict) -> InstanceResource:
"""Convert payload to resource instance"""
def get_instances(self) -> list:
"""Get all instances from page"""
def next_page_uri(self) -> str:
"""Get URI for next page"""
def previous_page_uri(self) -> str:
"""Get URI for previous page"""Exception Handling
Comprehensive exception hierarchy for error handling.
class TwilioException(Exception):
"""Base exception for all Twilio errors"""
def __init__(self, message: str):
self.message = message
super().__init__(message)
class TwilioRestException(TwilioException):
"""REST API error with detailed information"""
def __init__(
self,
status: int,
uri: str,
message: str = None,
code: int = None,
method: str = 'GET',
details: dict = None
):
"""
Args:
status (int): HTTP status code
uri (str): Request URI that failed
message (str): Error message
code (int): Twilio error code
method (str): HTTP method
details (dict): Additional error details
"""
self.status = status
self.uri = uri
self.code = code
self.method = method
self.details = details or {}
self.more_info = f"https://www.twilio.com/docs/errors/{code}" if code else None
super().__init__(message or f"HTTP {status} error")
class TwilioHttpException(TwilioException):
"""HTTP transport layer error"""
def __init__(self, message: str):
super().__init__(message)Utility Classes
Helper classes for parameter handling and data serialization.
class Values:
"""Parameter value container with special handling"""
def __init__(self, data: dict = None):
self._data = data or {}
def __getitem__(self, key: str):
return self._data.get(key)
def __setitem__(self, key: str, value):
if value is not None:
self._data[key] = value
def of(self, data: dict) -> 'Values':
"""Create Values instance from dict"""
return Values(data)
# Serialization utilities
def serialize_object(obj) -> str:
"""Serialize object to string representation"""
def serialize_list_object(obj_list: list, separator: str = ',') -> str:
"""Serialize list of objects to string"""
def deserialize_rfc2822_datetime(date_string: str) -> datetime:
"""Parse RFC2822 formatted datetime string"""
def deserialize_iso8601_datetime(date_string: str) -> datetime:
"""Parse ISO8601 formatted datetime string"""
def deserialize_integer(value: str) -> int:
"""Parse string to integer"""
def deserialize_decimal(value: str) -> Decimal:
"""Parse string to decimal"""
def deserialize_boolean(value: str) -> bool:
"""Parse string to boolean"""
def prefixed_collapsible_map(data: dict, prefix: str) -> dict:
"""Convert prefixed dictionary to collapsible format"""
def map_properties(data: dict) -> dict:
"""Map dictionary properties for API serialization"""Custom HTTP Client Example
from twilio.http.http_client import HttpClient
from twilio.http.response import Response
import requests
class CustomHttpClient(HttpClient):
"""Custom HTTP client with additional logging"""
def __init__(self, timeout=60, retries=3):
self.timeout = timeout
self.retries = retries
self.session = requests.Session()
def request(self, method, uri, data=None, headers=None, auth=None, **kwargs):
print(f"Making {method} request to {uri}")
response = self.session.request(
method=method,
url=uri,
data=data,
headers=headers,
auth=auth,
timeout=self.timeout,
**kwargs
)
print(f"Response status: {response.status_code}")
return Response(
status_code=response.status_code,
text=response.text,
headers=dict(response.headers)
)
# Use custom client
from twilio.rest import Client
custom_client = CustomHttpClient(timeout=30)
client = Client(
'ACxxxxx',
'auth_token',
http_client=custom_client
)Error Handling Patterns
from twilio.base.exceptions import TwilioRestException, TwilioException
try:
message = client.messages.create(
body="Test message",
from_="+15551234567",
to="invalid_number"
)
except TwilioRestException as e:
print(f"Twilio API Error:")
print(f" Status: {e.status}")
print(f" Code: {e.code}")
print(f" Message: {e.message}")
print(f" URI: {e.uri}")
print(f" More info: {e.more_info}")
# Handle specific error codes
if e.code == 21211:
print("Invalid 'To' phone number")
elif e.code == 21608:
print("Number not owned by account")
except TwilioException as e:
print(f"General Twilio error: {e.message}")
except Exception as e:
print(f"Unexpected error: {str(e)}")Install with Tessl CLI
npx tessl i tessl/pypi-twilio