sdk-creator 0.0.2

A dead simple asynchronous wrapper for RESTful APIs

SDK Creator

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.

Why SDK Creator?

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

Installation

pip install sdk-creator
# or using uv
uv add sdk-creator

Quick Start - Building Your First SDK

Here's how to build a clean, typed SDK wrapper around a Users API:

1. Define Your Models

# 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

2. Create Your SDK Class

# 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()

3. Use Your SDK

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="[e-mail]"
        )
        print(f"Created user: {new_user.user.name}")

asyncio.run(main())

Real-World Example: Azure Face API SDK

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

Key Features for SDK Development

🏗️ Composition Over Inheritance

• Use AsyncRestAdapter as 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

🔐 Flexible Authentication

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"}
        )

🛡️ Comprehensive Error Handling

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}")

📝 Strong Typing with Pydantic

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 = True

🧰 SDK Development Toolkit

SDK Creator includes a comprehensive toolkit of utilities to accelerate your SDK development:

URL and Endpoint Utilities

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"

Case Conversion Utilities

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"

Pydantic Model Enhancements

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=True

Exception Handling

from 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")

SDK Development Patterns

Environment Configuration

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})"}
        )

Pagination Support

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 += 1

Custom Exception Hierarchy

Create 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")

AsyncRestAdapter API Reference

Constructor Parameters

• hostname (str): API server hostname or full endpoint URL
• api_version (str): API version path (default: "v1")
• endpoint_prefix (str | None): Additional prefix for all endpoints
• api_key (str): API key for authentication
• ssl_verify (bool): Verify SSL certificates (default: True)
• scheme (Literal["http", "https"]): URL scheme (default: "https")
• jwt_token (str | None): JWT token for Bearer authentication
• azure_api (bool): Enable Azure API Management headers
• headers (dict | None): Additional default headers

HTTP Methods

All HTTP methods support an expect_json_response parameter for flexible response handling:

• get(endpoint, expect_json_response=True, **params) - GET request
• post(endpoint, data=None, expect_json_response=True, **params) - POST request
• put(endpoint, data=None, expect_json_response=False, **params) - PUT request
• patch(endpoint, data=None, expect_json_response=False, **params) - PATCH request
• delete(endpoint, data=None, expect_json_response=False, **params) - DELETE request

Exception Hierarchy

ApiError (base)
├── ApiRequestError        # Network/connection issues
├── ApiResponseError       # Response parsing errors
├── ApiTimeoutError        # Request timeouts
└── ApiRaisedFromStatusError  # HTTP error status codes

🧪 Quality & Testing

SDK Creator is built with production-grade quality standards:

Test Coverage

• 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)

Code Quality

• Zero linting errors with Ruff
• Full type annotations for excellent IDE support
• Professional error handling with detailed exception hierarchy
• Comprehensive documentation with practical examples

Development Tools

• 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

Testing Your SDK

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)

Best Practices

1. Keep SDKs Focused

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): ...

2. Use Composition

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):
    pass

3. Validate Input Early

Use 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)

ApiResponse

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 message

Exception Hierarchy

ApiError (base)
├── ApiRequestError        # Network/connection issues
├── ApiResponseError       # Response parsing errors
├── ApiTimeoutError        # Request timeouts
└── ApiRaisedFromStatusError  # HTTP error status codes

🚀 Development & Contributing

Development Setup

# 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

Quality Standards

• 100% test coverage maintained
• Zero linting errors with Ruff
• Full type annotations required
• Comprehensive tests for all functionality

Contributing

1. Fork the repository
2. Create a feature branch
3. Add tests for new functionality
4. Ensure 100% coverage is maintained
5. Submit a pull request

Roadmap

🚀 Next Release (v0.0.3)

• 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

🔮 Future Versions

• 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

License

This project is licensed under the MIT License.