Clients provide the low-level interface for connecting to external services, databases, and APIs.
ClientInterface
application_sdk.clients.ClientInterface
Base interface class for implementing client connections.
Methods
load
Establish the client connection.
async def load(*args: Any, **kwargs: Any) -> None
Variable positional arguments
Variable keyword arguments, typically including credentials
If the subclass does not implement this method
close
Close the client connection.
async def close(*args: Any, **kwargs: Any) -> None
Variable positional arguments
Variable keyword arguments
Default implementation does nothing. Override if cleanup is needed.
BaseClient
application_sdk.clients.base.BaseClient
Base client implementation for non-SQL based applications with HTTP support.
Constructor
BaseClient(
credentials: Dict[str, Any] = {},
http_headers: HeaderTypes = {}
)
credentials
Dict[str, Any]
default:"{}"
Client credentials for authentication
HTTP headers for all requests. Supports dict, Headers object, or list of tuples.
Attributes
Client credentials for authentication
HTTP headers for all http requests made by this client
HTTP transport for requests. Uses httpx default transport by default.
Can be overridden in load() method for custom retry behavior.
Methods
load
Initialize the client with credentials.
async def load(**kwargs: Any) -> None
Keyword arguments, typically including credentials
If the subclass does not implement this method
Subclasses should override this to:
- Set up authentication headers in self.http_headers
- Initialize any required client state
- Handle credential processing
- Optionally override self.http_retry_transport for custom retry behavior
execute_http_get_request
Perform an HTTP GET request.
await client.execute_http_get_request(
url: str,
headers: Optional[HeaderTypes] = None,
params: Optional[QueryParamTypes] = None,
auth: Optional[AuthTypes] = None,
timeout: int = 10
) -> Optional[httpx.Response]
The URL to make the GET request to
HTTP headers to include. These override/add to client-level headers.
Supports dict, Headers object, or list of tuples.
params
QueryParamTypes
default:"None"
Query parameters. Supports dict, list of tuples, or string.
Authentication to use. Supports BasicAuth, DigestAuth, custom auth classes, or tuples for basic auth.
Request timeout in seconds
The HTTP response if successful, None if failed
execute_http_post_request
Perform an HTTP POST request.
await client.execute_http_post_request(
url: str,
data: Optional[RequestData] = None,
json_data: Optional[Any] = None,
content: Optional[bytes] = None,
files: Optional[RequestFiles] = None,
headers: Optional[HeaderTypes] = None,
params: Optional[QueryParamTypes] = None,
cookies: Optional[Dict[str, str]] = None,
auth: Optional[AuthTypes] = None,
follow_redirects: bool = True,
verify: bool = True,
timeout: int = 30
) -> Optional[httpx.Response]
The URL to make the POST request to
data
RequestData
default:"None"
Form data to send. Supports dict, list of tuples, or other httpx-compatible formats.
JSON data to send. Any JSON-serializable object.
Raw binary content to send
files
RequestFiles
default:"None"
Files to upload. Supports various file formats and tuples.
HTTP headers to include. These override/add to client-level headers.
params
QueryParamTypes
default:"None"
Query parameters
cookies
Dict[str, str]
default:"None"
Cookies to include
Whether to follow HTTP redirects
Whether to verify SSL certificates
Request timeout in seconds
The HTTP response if successful, None if failed
Example Usage
Basic HTTP Client
from application_sdk.clients.base import BaseClient
from typing import Dict, Any, Optional
import httpx
class APIClient(BaseClient):
async def load(self, **kwargs):
"""Initialize client with API key."""
credentials = kwargs.get("credentials", {})
api_key = credentials.get("api_key")
# Set up authentication headers
self.http_headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
async def get_users(self) -> Optional[Dict[str, Any]]:
"""Fetch users from API."""
response = await self.execute_http_get_request(
url="https://api.example.com/users",
params={"limit": 100}
)
return response.json() if response else None
async def create_user(self, user_data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""Create a new user."""
response = await self.execute_http_post_request(
url="https://api.example.com/users",
json_data=user_data
)
return response.json() if response else None
Client with Retry Logic
from application_sdk.clients.base import BaseClient
from httpx_retries import Retry, RetryTransport
import httpx
class ResilientClient(BaseClient):
async def load(self, **kwargs):
credentials = kwargs.get("credentials", {})
# Set up headers
self.http_headers = {
"Authorization": f"Bearer {credentials.get('token')}"
}
# Configure retry transport for status code-based retries
retry = Retry(
total=5,
backoff_factor=10,
status_forcelist=[429, 500, 502, 503, 504]
)
self.http_retry_transport = RetryTransport(retry=retry)
async def fetch_data(self, endpoint: str):
"""Fetch data with automatic retries."""
response = await self.execute_http_get_request(
url=f"https://api.example.com/{endpoint}"
)
return response.json() if response else None
Client with Basic Auth
from application_sdk.clients.base import BaseClient
from httpx import BasicAuth
from typing import Optional, Dict, Any
class BasicAuthClient(BaseClient):
async def load(self, **kwargs):
credentials = kwargs.get("credentials", {})
self.username = credentials.get("username")
self.password = credentials.get("password")
async def fetch_protected_resource(self) -> Optional[Dict[str, Any]]:
"""Fetch resource using basic authentication."""
response = await self.execute_http_get_request(
url="https://api.example.com/protected",
auth=BasicAuth(self.username, self.password)
)
return response.json() if response else None
File Upload Client
from application_sdk.clients.base import BaseClient
from typing import Optional, Dict, Any
import os
class FileUploadClient(BaseClient):
async def load(self, **kwargs):
credentials = kwargs.get("credentials", {})
self.http_headers = {
"Authorization": f"Bearer {credentials.get('token')}"
}
async def upload_file(
self,
file_path: str,
description: str = ""
) -> Optional[Dict[str, Any]]:
"""Upload a file to the API."""
with open(file_path, "rb") as f:
file_content = f.read()
filename = os.path.basename(file_path)
response = await self.execute_http_post_request(
url="https://api.example.com/upload",
data={"description": description},
files={
"file": (filename, file_content, "application/octet-stream")
}
)
return response.json() if response else None
from application_sdk.clients.base import BaseClient
from httpx import Headers
from typing import Dict, Any
class MultiHeaderClient(BaseClient):
async def load(self, **kwargs):
credentials = kwargs.get("credentials", {})
# Set client-level headers using Headers object
self.http_headers = Headers({
"Authorization": f"Bearer {credentials.get('token')}",
"User-Agent": "MyApp/1.0",
"Accept": "application/json"
})
async def fetch_json(self, endpoint: str):
"""Fetch JSON data."""
response = await self.execute_http_get_request(
url=f"https://api.example.com/{endpoint}",
headers={"Content-Type": "application/json"} # Adds to client headers
)
return response.json() if response else None
async def fetch_xml(self, endpoint: str):
"""Fetch XML data."""
response = await self.execute_http_get_request(
url=f"https://api.example.com/{endpoint}",
headers={"Accept": "application/xml"} # Overrides client Accept header
)
return response.text if response else None
Client with Connection Pooling
from application_sdk.clients.base import BaseClient
import httpx
from typing import Optional, Dict, Any
class PooledClient(BaseClient):
def __init__(self):
super().__init__()
self._client: Optional[httpx.AsyncClient] = None
async def load(self, **kwargs):
credentials = kwargs.get("credentials", {})
self.http_headers = {
"Authorization": f"Bearer {credentials.get('token')}"
}
# Create persistent client with connection pooling
self._client = httpx.AsyncClient(
headers=self.http_headers,
timeout=30.0,
limits=httpx.Limits(
max_keepalive_connections=20,
max_connections=100
)
)
async def fetch_data(self, endpoint: str) -> Optional[Dict[str, Any]]:
"""Fetch data using pooled connection."""
if not self._client:
raise ValueError("Client not loaded")
response = await self._client.get(
f"https://api.example.com/{endpoint}"
)
return response.json() if response.status_code == 200 else None
async def close(self):
"""Close the client and cleanup connections."""
if self._client:
await self._client.aclose()
self._client = None
Best Practices
Client Design
- Keep clients focused on transport and low-level operations
- Implement connection management in load() and close()
- Use connection pooling for high-throughput scenarios
- Handle authentication in load() method
HTTP Operations
- Use client-level headers for authentication
- Use method-level headers for request-specific settings
- Configure appropriate timeouts
- Use retry logic for unreliable services
Error Handling
- Return None on errors for graceful degradation
- Log errors with appropriate context
- Close connections in finally blocks
- Implement close() method for cleanup
- Reuse client instances
- Use connection pooling
- Configure appropriate timeouts
- Consider async context managers for resource management