pydantic-client

Pydantic Client

A flexible HTTP client library that leverages Pydantic models for request/response handling, supporting both synchronous and asynchronous operations.

⭐ If you like this project, please star it on GitHub!

Features

• 🔥 Type-safe: Full type hints with Pydantic models for request/response validation
• 🚀 Multiple HTTP backends: Choose from requests, aiohttp, or httpx
• ⚡ Async/Sync support: Work with both synchronous and asynchronous HTTP operations
• 🎯 Decorator-based API: Clean, intuitive API definition with decorators
• 📝 CLI tools: Command-line interface for automatic client generation from OpenAPI/Swagger specs
• 🛡️ Mock API Responses: This is useful for testing or development purposes.
• ⚡ Timing context manager: Use with client.span(prefix="myapi"): to log timing for any API call, sync or async
• 🌟 Nested Response Extraction: Extract and parse deeply nested API responses using JSON path expressions

Installation

pip install pydantic-client

Examples

See the example/ directory for real-world usage of this library, including:

• example_requests.py: Synchronous usage with RequestsWebClient
• example_httpx.py: Async usage with HttpxWebClient
• example_aiohttp.py: Async usage with AiohttpWebClient
• example_tools.py: How to register and use Agno tools
• example_nested_response.py: How to extract data from nested API responses

Quick Start

from pydantic import BaseModel
from pydantic_client import RequestsWebClient, get, post


# Define your response models
class UserResponse(BaseModel):
    id: int
    name: str
    email: str


class CreateUser(BaseModel):
    name: str
    email: str


# Create your API client
class MyAPIClient(RequestsWebClient):
    def __init__(self):
        super().__init__(
            base_url="https://api.example.com",
            headers={"Authorization": "Bearer token"}
        )

    @get("users/{user_id}")
    def get_user(self, user_id: int) -> UserResponse:
        pass

    @post("users")
    def create_user(self, user: CreateUser) -> UserResponse:
        pass

    @get("/users/string?name={name}")
    def get_user_string(self, name: Optional[str] = None) -> dict:
        # will get raw json data
        ...
    
    @get("/users/{user_id}/bytes")
    def get_user_bytes(self, user_id: str) -> bytes:
        # will get raw content, bytes type.
        ...

    @delete(
        "/users",
        agno_tool=True,
        tool_description="description or use function annotation."
    )
    def delete_user(self, user_id: str, request_headers: Dict[str, Any]):
        ...


# Use the client
client = MyAPIClient(base_url="https://localhost")
user = client.get_user(user_id=123)

user_body = CreateUser(name="john", email="[e-mail]")
user = client.create_user(user_body)

# will update the client headers.
client.delete_user("123", {"ba": "your"})


from agno.agent import Agent

agent = Agent(.....)
client.register_agno_tools(agent)  # delete_user is used by tools.

Available Clients

• RequestsWebClient: Synchronous client using the requests library
• AiohttpWebClient: Asynchronous client using aiohttp
• HttpxWebClient: Asynchronous client using httpx

HTTP Method Decorators

The library provides decorators for common HTTP methods:

• @get(path)
• @post(path)
• @put(path)
• @patch(path)
• @delete(path)

Path Parameters

Path parameters are automatically extracted from the URL template and matched with method arguments:

@get("users/{user_id}/posts/{post_id}")
def get_user_post(self, user_id: int, post_id: int) -> PostResponse:
    pass

Request Parameters

• For GET and DELETE methods, remaining arguments are sent as query parameters
• For POST, PUT, and PATCH methods, remaining arguments are sent in the request body as JSON

# you can call signature by your self, overwrite the function `before_request`


class MyAPIClient(RequestsWebClient):
    # some code

    def before_request(self, request_params: Dict[str, Any]) -> Dict[str, Any]:
        # the request params before send: body, header, etc...
        sign = cal_signature(request_params)
        request_params["headers"].update(dict(signature=sign))
        return request_params


# will send your new request_params
user = client.get_user("123")

Handling Nested API Responses

Many APIs return deeply nested JSON structures. Use the response_extract_path parameter to extract and parse specific data from complex API responses:

from typing import List
from pydantic import BaseModel
from pydantic_client import RequestsWebClient, get

class User(BaseModel):
    id: str
    name: str
    email: str

class MyClient(RequestsWebClient):
    @get("/users/complex", response_extract_path="$.data.users")
    def get_users_nested(self) -> List[User]:
        """
        Extracts the users list from a nested response structure
        
        Example response:
        {
            "status": "success",
            "data": {
                "users": [
                    {"id": "1", "name": "Alice", "email": "[e-mail]"},
                    {"id": "2", "name": "Bob", "email": "[e-mail]"}
                ],
                "total": 2
            }
        }
        """
        pass
    
    @get("/search", response_extract_path="$.results[0]")
    def search_first_result(self) -> User:
        """
        Get just the first search result from an array
        """
        pass

The response_extract_path parameter defines where to find the data in the response. It supports:

• Array indexing with square brackets: $.results[0] -> User
• Optional $ prefix for root object: $.data.users -> list[User]

Mock API Responses

You can configure the client to return mock responses instead of making actual API calls. This is useful for testing or development purposes.

Setting Mock Responses Directly

from pydantic_client import RequestsWebClient, get


class UserResponse(BaseModel):
    id: int
    name: str


class MyClient(RequestsWebClient):
    @get("/users/{user_id}")
    def get_user(self, user_id: int) -> UserResponse:
        pass

# Create client and configure mocks
client = MyClient(base_url="https://api.example.com")
client.set_mock_config(mock_config=[
    {
        "name": "get_user",
        "output": {
            "id": 123,
            "name": "Mock User"
        }
    }
])

# This will return the mock response without making an actual API call
user = client.get_user(1)  # Returns UserResponse(id=123, name="Mock User")

Loading Mock Responses from a JSON File

You can also load mock configurations from a JSON file:

# Load mock data from a JSON file
client.set_mock_config(mock_config_path="path/to/mock_data.json")

The JSON file should follow this format:

[
    {
        "name": "get_user",
        "output": {
            "id": 123,
            "name": "Mock User"
        }
    },
    {
        "name": "list_users",
        "output": {
            "users": [
                {"id": 1, "name": "User 1"},
                {"id": 2, "name": "User 2"}
            ]
        }
    }
]

Setting Mock Responses in Client Configuration

You can also include mock configuration when creating a client from configuration:

config = {
    "base_url": "https://api.example.com",
    "timeout": 10,
    "mock_config": [
        {
            "name": "get_user",
            "output": {
                "id": 123,
                "name": "Mock User"
            }
        }
    ]
}

client = MyClient.from_config(config)

CLI Tools

Generate a client from an OpenAPI/Swagger specification:

# Generate a requests client from a YAML file
swagger-cli -f api.yaml -t requests -o api_client.py

# Generate an aiohttp client from a JSON file
swagger-cli -f openapi.json -t aiohttp -o async_api_client.py

# Generate an httpx client with custom output name
swagger-cli -f swagger.yaml -t httpx -o http_client.py

CLI Features

The CLI tool automatically handles common OpenAPI/Swagger specification issues:

• Python Keyword Conflicts: Parameter/field names that conflict with Python keywords (e.g., import, from, class) are automatically renamed with an underscore suffix (e.g., import_, from_, class_)

• Invalid Python Identifiers: Parameter/field names with invalid characters (e.g., hyphens, dots, special characters) are automatically converted to valid Python identifiers using underscores:

  • x-immich-checksum → x_immich_checksum
  • api-key → api_key
  • user.id → user_id

• Parameter Ordering: Required parameters (without default values) are automatically placed before optional parameters (with default values), regardless of their order in the OpenAPI specification. This ensures generated code follows Python's syntax rules.

Example generated code:

# Original OpenAPI spec with mixed parameter order and invalid identifiers
@put("/albums/{id}/assets")
async def add_assets_to_album(
    self,
    id: str,
    bulkidsdto: BulkIdsDto,  # Required - moved before optional params
    key: str | None = None,  # Optional
    slug: str | None = None  # Optional
):
    """Upload asset"""
    ...

Timing Context Manager

All clients support a span context manager for simple API call timing and logging:

with client.span(prefix="fetch_user"):
    user = client.get_user_by_id("123")
# Logs the elapsed time for the API call, useful for performance monitoring.

# will send `fetch_user.elapsed` to statsd
client = MyAPIClient(base_url="https://localhost", statsd_address="localhost:8125")
with client.span(prefix="fetch_user"):
    user = client.get_user_by_id("123")

Custom Configuration

You can initialize clients with custom configurations:

client = MyAPIClient(
    base_url="https://api.example.com",
    headers={"Custom-Header": "value"},
    session=requests.Session(),  # your own session
    timeout=30  # in seconds
)

# Or using configuration dictionary
config = {
    "base_url": "https://api.example.com",
    "headers": {"Custom-Header": "value"},
    "timeout": 30
}
client = MyAPIClient.from_config(config)