A foundation for building strongly-typed Python SDKs around existing REST APIs. SDK Creator provides the async HTTP foundation, comprehensive utilities, and production-grade testing while you focus on building clean, Pydantic-powered API wrappers with comprehensive error handling and type safety.
Instead of manually handling HTTP requests, JSON parsing, and error handling for every API integration, SDK Creator lets you:
- Build clean SDK interfaces with strong typing and Pydantic models
- Focus on business logic rather than HTTP boilerplate
- Leverage async/await for high-performance API calls
- Handle errors gracefully with comprehensive exception types
- Maintain consistency across multiple API integrations
- Use battle-tested utilities for common SDK development tasks
- Benefit from 100% test coverage and production-ready code quality
pip install sdk-creator
# or using uv
uv add sdk-creatorHere's how to build a clean, typed SDK wrapper around a Users API:
# models/responses.py
from sdk_creator.toolkit import SdkModel, CamelCaseAliasMixin
from typing import List, Optional
class User(CamelCaseAliasMixin):
"""User model with automatic camelCase API field mapping."""
user_id: int # Maps to "userId" in API
full_name: str # Maps to "fullName" in API
email_address: str # Maps to "emailAddress" in API
is_active: bool # Maps to "isActive" in API
class UserList(SdkModel):
"""User list response with enhanced SDK model features."""
users: List[User]
total_count: int # Maps to "totalCount" in API
current_page: int # Maps to "currentPage" in API
class CreateUserResponse(BaseModel):
user: User
message: str# users_sdk.py
from typing import Optional, Self, Any
from sdk_creator import AsyncRestAdapter
from sdk_creator.errors import ApiRaisedFromStatusError
from sdk_creator.toolkit import SdkModel, CamelCaseAliasMixin, join_endpoints
from .models.responses import User, UserList, CreateUserResponse
class UsersSDK:
def __init__(self, api_key: str, base_url: str = "api.example.com"):
"""Initialize the Users SDK.
Args:
api_key: Your API key for authentication
base_url: API hostname (default: api.example.com)
"""
self._adapter = AsyncRestAdapter(
hostname=base_url,
api_version="v1",
endpoint_prefix="users", # all endpoints will be prefixed with "users"
api_key=api_key,
scheme="https"
)
async def get_users(self, page: int = 1, limit: int = 10) -> UserList:
"""Get paginated list of users."""
response = await self._adapter.get("users", page=page, limit=limit)
return UserList.model_validate(response.data)
async def get_user(self, user_id: int) -> Optional[User]:
"""Get a specific user by ID."""
try:
response = await self._adapter.get(f"users/{user_id}")
return User.model_validate(response.data)
except ApiRaisedFromStatusError as e:
if e.status_code == 404:
return None
raise
async def create_user(self, name: str, email: str) -> CreateUserResponse:
"""Create a new user."""
data = {"name": name, "email": email}
response = await self._adapter.post("users", data=data)
return CreateUserResponse.model_validate(response.data)
async def update_user(self, user_id: int, **updates) -> User:
"""Update user information."""
response = await self._adapter.patch(f"users/{user_id}", data=updates)
return User.model_validate(response.data)
async def delete_user(self, user_id: int) -> bool:
"""Delete a user."""
try:
await self._adapter.delete(f"users/{user_id}")
return True
except ApiRaisedFromStatusError as e:
if e.status_code == 404:
return False
raise
async def close(self):
"""Close the HTTP client."""
await self._adapter.close()
async def __aenter__(self) -> Self:
return self
async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
await self.close()import asyncio
from users_sdk import UsersSDK
async def main():
async with UsersSDK(api_key="your-api-key") as sdk:
# Get users with strong typing
users = await sdk.get_users(page=1, limit=5)
print(f"Found {users.total} users")
# Get specific user (handles 404 gracefully)
user = await sdk.get_user(123)
if user:
print(f"User: {user.name} ({user.email})")
# Create new user
new_user = await sdk.create_user(
name="John Doe",
email="john@example.com"
)
print(f"Created user: {new_user.user.name}")
asyncio.run(main())Here's how SDK Creator is used to build a production-ready Azure Face API wrapper:
class PersonDirectory:
def __init__(self, azure_ai_endpoint: str, api_key: str):
hostname = self._extract_hostname(azure_ai_endpoint) + "/face"
self._adapter = AsyncRestAdapter(
hostname=hostname,
api_version="v1.2-preview.1",
api_key=api_key,
scheme="https"
)
async def get_persons(self, start: str | None = None, top: int = 10) -> PersonDirectoryPersons:
"""List all persons with strong typing and validation."""
response = await self._adapter.get("persons", start=start, top=top)
return PersonDirectoryPersons.model_validate({"persons": response.data})
async def create_person(self, name: str, user_data: str | dict) -> CreatePersonResult:
"""Create person with automatic JSON serialization."""
if isinstance(user_data, dict):
user_data = json.dumps(user_data)
person_data = PersonDirectoryCreate(name=name, user_data=user_data)
response = await self._adapter.post("persons", data=person_data.model_dump())
return CreatePersonResult.model_validate(response.data)
async def delete_person(self, person_id: str, *, raise_not_found: bool = True) -> bool:
"""Delete person with graceful 404 handling."""
try:
await self._adapter.delete(f"persons/{person_id}")
return True
except ApiRaisedFromStatusError as err:
if err.status_code == 404 and not raise_not_found:
return False
raise PersonDirectoryNotFoundError(f"Person {person_id} not found") from err- Use
AsyncRestAdapteras a private component in your SDK classes - Build clean, domain-specific interfaces on top of HTTP operations
- Maintain separation between transport logic and business logic
Configure authentication once in your SDK constructor:
class MySDK:
def __init__(self, api_key: str, environment: str = "production"):
base_urls = {
"production": "api.example.com",
"staging": "staging-api.example.com"
}
self._adapter = AsyncRestAdapter(
hostname=base_urls[environment],
api_key=api_key,
headers={"User-Agent": "MySDK/1.0"}
)Transform HTTP errors into meaningful domain exceptions with a complete error hierarchy:
from sdk_creator.errors import (
ApiRaisedFromStatusError, ApiTimeoutError,
ApiRequestError, ApiResponseError
)
class UserNotFoundError(Exception):
pass
class UserSDK:
async def get_user(self, user_id: int) -> User:
try:
response = await self._adapter.get(f"users/{user_id}")
return User.model_validate(response.data)
except ApiRaisedFromStatusError as e:
if e.status_code == 404:
raise UserNotFoundError(f"User {user_id} not found") from e
raise # Re-raise other HTTP errors
except ApiTimeoutError:
raise TimeoutError("User service unavailable")
except ApiRequestError as e:
raise ConnectionError(f"Network error: {e}")
except ApiResponseError as e:
raise ValueError(f"Invalid response format: {e}")Automatic validation and serialization of API responses:
from pydantic import BaseModel, Field
from datetime import datetime
class User(BaseModel):
id: int
email: str = Field(..., pattern=r'^[^@]+@[^@]+\.[^@]+$')
created_at: datetime
is_active: bool = True
class Config:
# Automatically convert API snake_case to Python snake_case
allow_population_by_field_name = TrueSDK Creator includes a comprehensive toolkit of utilities to accelerate your SDK development:
from sdk_creator.toolkit import url_to_hostname, join_endpoints
# extract hostname from URLs with validation
hostname = url_to_hostname("https://api.example.com/v1/users")
# returns: "api.example.com"
# intelligently join URL endpoints
endpoint = join_endpoints("api", "v1", "users", "123")
# returns: "api/v1/users/123"
# handles edge cases gracefully
endpoint = join_endpoints("/api/", "//v1//", "/users/")
# returns: "api/v1/users"from sdk_creator.toolkit import to_camelcase
# convert snake_case to camelCase for API compatibility
camel_field = to_camelcase("user_id") # Returns: "userId"
camel_field = to_camelcase("created_at_timestamp") # Returns: "createdAtTimestamp"from sdk_creator.toolkit import CamelCaseAliasMixin, SdkModel
# automatic camelCase field aliases for APIs
class UserModel(CamelCaseAliasMixin):
user_id: int # API field: "userId"
created_at: str # API field: "createdAt"
is_active: bool # API field: "isActive"
# enhanced base model with SDK-specific configurations
class ApiResponse(SdkModel):
data: dict
status: str
# inherits from_attributes=True and arbitrary_types_allowed=Truefrom sdk_creator.errors import (
ApiError, ApiRequestError, ApiResponseError,
ApiTimeoutError, ApiRaisedFromStatusError
)
try:
response = await sdk.get_user(123)
except ApiRaisedFromStatusError as e:
print(f"HTTP {e.status_code} error: {e}")
except ApiTimeoutError:
print("Request timed out")
except ApiRequestError:
print("Network connection failed")
except ApiResponseError:
print("Invalid response format")
except ApiError:
print("General API error")Support multiple environments in your SDK:
class MySDK:
ENVIRONMENTS = {
"production": "api.example.com",
"staging": "staging-api.example.com",
"development": "dev-api.example.com"
}
def __init__(self, api_key: str, environment: str = "production"):
if environment not in self.ENVIRONMENTS:
raise ValueError(f"Invalid environment: {environment}")
self._adapter = AsyncRestAdapter(
hostname=self.ENVIRONMENTS[environment],
api_key=api_key,
headers={"User-Agent": f"MySDK/1.0 ({environment})"}
)Handle paginated responses cleanly:
from typing import AsyncIterator
class MySDK:
async def get_all_users(self) -> AsyncIterator[User]:
"""Stream all users across multiple pages."""
page = 1
while True:
response = await self._adapter.get("users", page=page, limit=100)
user_data = UserPage.model_validate(response.data)
for user in user_data.users:
yield user
if not user_data.has_next:
break
page += 1Create meaningful exceptions for your domain:
class MySDKError(Exception):
"""Base exception for MySDK operations."""
class ValidationError(MySDKError):
"""Invalid input data."""
class ResourceNotFoundError(MySDKError):
"""Requested resource not found."""
class RateLimitError(MySDKError):
"""Rate limit exceeded."""
def __init__(self, retry_after: int):
self.retry_after = retry_after
super().__init__(f"Rate limited. Retry after {retry_after} seconds")hostname(str): API server hostname or full endpoint URLapi_version(str): API version path (default: "v1")endpoint_prefix(str | None): Additional prefix for all endpointsapi_key(str): API key for authenticationssl_verify(bool): Verify SSL certificates (default: True)scheme(Literal["http", "https"]): URL scheme (default: "https")jwt_token(str | None): JWT token for Bearer authenticationazure_api(bool): Enable Azure API Management headersheaders(dict | None): Additional default headers
All HTTP methods support an expect_json_response parameter for flexible response handling:
get(endpoint, expect_json_response=True, **params)- GET requestpost(endpoint, data=None, expect_json_response=True, **params)- POST requestput(endpoint, data=None, expect_json_response=False, **params)- PUT requestpatch(endpoint, data=None, expect_json_response=False, **params)- PATCH requestdelete(endpoint, data=None, expect_json_response=False, **params)- DELETE request
ApiError (base)
├── ApiRequestError # Network/connection issues
├── ApiResponseError # Response parsing errors
├── ApiTimeoutError # Request timeouts
└── ApiRaisedFromStatusError # HTTP error status codes
SDK Creator is built with production-grade quality standards:
- 100% source code coverage across all modules
- 162 comprehensive tests covering all functionality
- Continuous integration with GitHub Actions
- Multi-Python version support (3.11, 3.12, 3.13)
- Zero linting errors with Ruff
- Full type annotations for excellent IDE support
- Professional error handling with detailed exception hierarchy
- Comprehensive documentation with practical examples
- Automated coverage reporting with badge integration
- Pre-commit hooks for code formatting and linting
- Development scripts for testing and coverage verification
- CI/CD workflows for automated quality checks
SDK Creator provides excellent testing foundations for your own SDKs:
import pytest
from unittest.mock import Mock
from your_sdk import UserSDK
from sdk_creator.errors import ApiRaisedFromStatusError
@pytest.mark.asyncio
async def test_user_not_found():
sdk = UserSDK("test-key")
# mock the adapter to return 404
mock_response = Mock()
mock_response.status_code = 404
sdk._adapter._client.request = Mock(side_effect=ApiRaisedFromStatusError(404, "Not Found"))
with pytest.raises(UserNotFoundError):
await sdk.get_user(999)Create separate SDK classes for different API domains:
# ✅ Good - focused SDKs
class UsersSDK: ...
class OrdersSDK: ...
class PaymentsSDK: ...
# ❌ Avoid - monolithic SDK
class MegaSDK:
def get_user(self): ...
def create_order(self): ...
def process_payment(self): ...Keep AsyncRestAdapter as a private implementation detail:
# ✅ Good - adapter is private
class MySDK:
def __init__(self, api_key: str):
self._adapter = AsyncRestAdapter(...)
# ❌ Avoid - exposing internals
class MySDK(AsyncRestAdapter):
passUse Pydantic models for request validation:
class CreateUserRequest(BaseModel):
name: str = Field(..., min_length=1, max_length=100)
email: str = Field(..., pattern=r'^[^@]+@[^@]+\.[^@]+$')
age: int = Field(..., ge=0, le=150)
async def create_user(self, request: CreateUserRequest) -> User:
# validation happens automatically
response = await self._adapter.post("users", data=request.model_dump())
return User.model_validate(response.data)Response object returned by all HTTP methods:
from pydantic import BaseModel
class ApiResponse(BaseModel):
status_code: int # HTTP status code
data: Json # Parsed response data
message: str | None # Status messageApiError (base)
├── ApiRequestError # Network/connection issues
├── ApiResponseError # Response parsing errors
├── ApiTimeoutError # Request timeouts
└── ApiRaisedFromStatusError # HTTP error status codes
# clone and setup development environment
git clone https://github.com/mghalix/sdk-creator.git
cd sdk-creator
# install with development dependencies
uv sync --dev
# run tests with coverage
python scripts/check_coverage.py
# format and lint code
python -m ruff format .
python -m ruff check .
# or
./scripts/lint.sh- 100% test coverage maintained
- Zero linting errors with Ruff
- Full type annotations required
- Comprehensive tests for all functionality
- Fork the repository
- Create a feature branch
- Add tests for new functionality
- Ensure 100% coverage is maintained
- Submit a pull request
- Built-in Caching - Response caching with TTL, Redis/memory backends
- Rate Limiting - Automatic rate limiting with exponential backoff
- Enhanced Pagination - Auto-pagination with generators and cursor support
- Mock Server - Built-in testing utilities with mock responses
- Circuit Breaker - Fault tolerance patterns for resilient SDKs
- Metrics & Monitoring - Request/response metrics and health checks
- OpenAPI Integration - Auto-generate SDKs from OpenAPI specs
This project is licensed under the MIT License.