Python - 💧 Mizu

The Python SDK generator creates modern Python client libraries from your contract definitions. The generated code features full type hints, dataclass models, and both synchronous and asynchronous clients powered by httpx.

Key Features

Modern Python: Python 3.8+ with full type hints
Sync and Async: Both Client (sync) and AsyncClient (async) clients
httpx Powered: Industry-standard HTTP client with automatic retries
Dataclass Models: Clean, typed data structures
uv-Ready: Includes pyproject.toml for modern Python packaging
SSE Streaming: First-class Server-Sent Events support

Quick Start

Step 1: Define Your Contract

# api.yaml
name: TodoAPI
description: Todo list API

defaults:
  base_url: http://localhost:8080

resources:
  - name: todos
    description: Manage todo items
    methods:
      - name: create
        input: CreateInput
        output: Todo
        http:
          method: POST
          path: /todos
      - name: list
        output: ListOutput
        http:
          method: GET
          path: /todos
      - name: get
        input: GetInput
        output: Todo
        http:
          method: GET
          path: /todos/{id}
      - name: delete
        input: DeleteInput
        http:
          method: DELETE
          path: /todos/{id}

types:
  - name: Todo
    kind: struct
    fields:
      - name: id
        type: string
      - name: title
        type: string
      - name: done
        type: bool
  - name: CreateInput
    kind: struct
    fields:
      - name: title
        type: string
  - name: GetInput
    kind: struct
    fields:
      - name: id
        type: string
  - name: DeleteInput
    kind: struct
    fields:
      - name: id
        type: string
  - name: ListOutput
    kind: struct
    fields:
      - name: items
        type: "[]Todo"
      - name: total
        type: int

Step 2: Generate the SDK

mizu contract gen api.yaml --client --lang python --output ./sdk/python --package todoclient --version 1.0.0

Step 3: Install and Use

# Install the generated SDK
cd sdk/python
uv pip install -e .

# Or with pip
pip install -e .

from todoclient import Client

# Create a client
client = Client(base_url="http://localhost:8080")

# Create a todo
todo = client.todos.create(title="Learn Mizu Python SDK")
print(f"Created: {todo.id}")

# List all todos
result = client.todos.list()
print(f"Total: {result.total} todos")

# Get a specific todo
todo = client.todos.get(id=todo.id)
print(f"Got: {todo.id} - {todo.title}")

# Delete
client.todos.delete(id=todo.id)
print("Deleted successfully")

Generated Code Structure

sdk/python/
├── pyproject.toml
└── src/
    └── todoclient/
        ├── __init__.py
        ├── _client.py
        ├── _types.py
        ├── _resource.py
        └── _streaming.py

pyproject.toml

[project]
name = "todoclient"
version = "1.0.0"
requires-python = ">=3.8"
dependencies = ["httpx>=0.24.0"]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

init.py

from ._client import Client, AsyncClient

__all__ = ["Client", "AsyncClient"]

Client Configuration

Creating a Client

from todoclient import Client

# Basic client
client = Client(base_url="http://localhost:8080")

# With authentication
client = Client(
    api_key="your-api-key",
    base_url="http://localhost:8080",
)

# With all options
client = Client(
    api_key="your-api-key",
    base_url="http://localhost:8080",
    timeout=30.0,
    max_retries=3,
    default_headers={"X-Custom-Header": "value"},
)

Configuration Options

Option	Type	Default	Description
`api_key`	`str \| None`	`None`	Authentication token
`base_url`	`str \| None`	Service default	API base URL
`timeout`	`float \| None`	`None` (no timeout)	Request timeout in seconds
`max_retries`	`int`	`2`	Maximum retry attempts
`default_headers`	`Dict[str, str] \| None`	`None`	Headers for all requests

Modifying Client Options

Create a new client with modified options:

# Original client
client = Client(api_key="key-1", base_url="http://localhost:8080")

# New client with different API key
other_client = client.with_options(api_key="key-2")

# New client with additional headers
debug_client = client.with_options(
    default_headers={"X-Debug": "true"}
)

Closing the Client

Always close the client when done:

client = Client(base_url="http://localhost:8080")
try:
    # Use client...
    pass
finally:
    client.close()

Or use a context manager pattern:

from contextlib import closing

with closing(Client(base_url="http://localhost:8080")) as client:
    todo = client.todos.create(title="Task")

Async Client

The SDK includes a fully async client for use with asyncio:

import asyncio
from todoclient import AsyncClient

async def main():
    client = AsyncClient(base_url="http://localhost:8080")

    try:
        # Create a todo
        todo = await client.todos.create(title="Async task")
        print(f"Created: {todo.id}")

        # List todos
        result = await client.todos.list()
        for t in result.items:
            print(f"- {t.title}")

    finally:
        await client.close()

asyncio.run(main())

Async Context Manager

async def main():
    client = AsyncClient(base_url="http://localhost:8080")
    try:
        todos = await client.todos.list()
    finally:
        await client.aclose()  # or await client.close()

Type System

Type Mapping Reference

Contract Type	Python Type
`string`	`str`
`bool`	`bool`
`int`, `int8`-`int64`	`int`
`uint`, `uint8`-`uint64`	`int`
`float32`, `float64`	`float`
`time.Time`	`datetime`
`json.RawMessage`	`object`
`any`	`object`
`[]T`	`List[T]`
`map[string]T`	`Dict[str, T]`

Dataclass Types

Contract struct types generate Python dataclasses:

from dataclasses import dataclass
from typing import Optional
from datetime import datetime

@dataclass
class Todo:
    id: str
    title: str
    done: bool
    created_at: datetime

Optional and Nullable Fields

Contract Definition	Python Type
Required field	`T`
`optional: true`	`Optional[T]`
`nullable: true`	`Optional[T]`
Both optional and nullable	`Optional[T]`

Example:

@dataclass
class UpdateInput:
    title: str                    # required
    done: Optional[bool] = None   # optional

Enum Fields

Enum fields use literal type hints in the docstring:

@dataclass
class Task:
    # status: One of "pending", "active", "completed"
    status: str

List and Dict Types

from typing import List, Dict

# Contract: kind: slice, elem: Todo
TodoList = List[Todo]

# Contract: kind: map, elem: string
Metadata = Dict[str, str]

Resources and Methods

Resource Pattern

Each contract resource becomes a property on the client:

client.todos      # TodosResource
client.users      # UsersResource

Method Signatures

Methods use Python naming conventions (snake_case):

Contract Method	Python Signature
`Create`	`create(**kwargs) -> Todo`
`List`	`list() -> ListOutput`
`Get`	`get(id: str) -> Todo`
`Delete`	`delete(id: str) -> None`

Calling Methods

Methods accept keyword arguments matching the input type fields:

# If input type is CreateInput { title: string }
todo = client.todos.create(title="My task")

# If input type is UpdateInput { title: string, done: bool (optional) }
todo = client.todos.update(id="123", title="Updated", done=True)

Streaming (SSE)

For methods with streaming support, the SDK provides iterator-based consumption:

Sync Streaming

# Stream events
for event in client.responses.stream(model="gpt-4", input="Hello"):
    if event.type == "text_delta":
        print(event.text, end="", flush=True)
    elif event.type == "completed":
        print("\n--- Done ---")

Async Streaming

async for event in client.responses.stream(model="gpt-4", input="Hello"):
    if event.type == "text_delta":
        print(event.text, end="", flush=True)

Collecting All Events

# Collect all events into a list
events = list(client.responses.stream(model="gpt-4", input="Hello"))

# Or async
events = [event async for event in client.responses.stream(...)]

Error Handling

Error Types

The SDK defines three error types:

class SDKError(Exception):
    """Base exception for all SDK errors."""
    pass

class APIConnectionError(SDKError):
    """Raised when unable to connect to the API."""
    pass

class APIStatusError(SDKError):
    """Raised when the API returns an error status code."""
    status_code: int
    body: Any

Handling Errors

from todoclient import Client
from todoclient._client import APIStatusError, APIConnectionError

client = Client(base_url="http://localhost:8080")

try:
    todo = client.todos.get(id="nonexistent")
except APIStatusError as e:
    if e.status_code == 404:
        print("Todo not found")
    elif e.status_code == 401:
        print("Unauthorized")
    elif e.status_code == 400:
        print(f"Bad request: {e}")
    else:
        print(f"API error {e.status_code}: {e}")
except APIConnectionError as e:
    print(f"Connection failed: {e}")

Automatic Retries

The client automatically retries failed requests (except for 4xx errors):

# Configure max retries
client = Client(
    base_url="http://localhost:8080",
    max_retries=5,  # Retry up to 5 times
)

Retries are attempted for:

Network errors
5xx server errors
Timeout errors

Advanced Usage

Custom Headers

# Default headers for all requests
client = Client(
    base_url="http://localhost:8080",
    default_headers={
        "X-Request-ID": "abc-123",
        "X-API-Version": "2024-01",
    },
)

Authentication Modes

The SDK supports different authentication modes based on your service configuration:

# Bearer token (default)
client = Client(api_key="your-token")
# Sends: Authorization: Bearer your-token

# Basic auth (if service configured with auth: basic)
client = Client(api_key="base64-encoded-credentials")
# Sends: Authorization: Basic base64-encoded-credentials

# No auth (if service configured with auth: none)
client = Client()
# Sends no Authorization header

Timeout Configuration

# Set timeout in seconds
client = Client(
    base_url="http://localhost:8080",
    timeout=30.0,  # 30 second timeout
)

# No timeout (wait indefinitely)
client = Client(
    base_url="http://localhost:8080",
    timeout=None,
)

Using with Existing httpx Client

The generated SDK uses httpx internally. For advanced use cases, you can access the underlying client:

# The SDK creates its own httpx client
client = Client(base_url="http://localhost:8080")
# client._http is the httpx.Client instance

Complete Example

Server

// main.go
package main

import (
    "github.com/go-mizu/mizu"
    contract "github.com/go-mizu/mizu/contract/v2"
    "github.com/go-mizu/mizu/contract/v2/transport/rest"

    "yourapp/todo"
)

func main() {
    impl := todo.NewService()
    svc := contract.Register[todo.API](impl,
        contract.WithDefaultResource("todos"),
        contract.WithDefaults(&contract.Defaults{
            BaseURL: "http://localhost:8080",
        }),
    )

    app := mizu.New()
    rest.Mount(app.Router, svc)
    app.Listen(":8080")
}

Client Usage

#!/usr/bin/env python3
"""Example usage of the generated Todo client."""

from todoclient import Client

def main():
    # Create client
    client = Client(base_url="http://localhost:8080")

    try:
        # Create a todo
        todo = client.todos.create(title="Buy groceries")
        print(f"Created: {todo.id}")

        # List all todos
        result = client.todos.list()
        print(f"\nAll todos ({result.total} total):")
        for t in result.items:
            status = "x" if t.done else " "
            print(f"  [{status}] {t.id}: {t.title}")

        # Get a specific todo
        fetched = client.todos.get(id=todo.id)
        print(f"\nFetched: {fetched.title}")

        # Delete the todo
        client.todos.delete(id=todo.id)
        print(f"\nDeleted: {todo.id}")

    finally:
        client.close()

if __name__ == "__main__":
    main()

Async Example

#!/usr/bin/env python3
"""Async example usage of the generated Todo client."""

import asyncio
from todoclient import AsyncClient

async def main():
    client = AsyncClient(base_url="http://localhost:8080")

    try:
        # Create multiple todos concurrently
        tasks = [
            client.todos.create(title="Task 1"),
            client.todos.create(title="Task 2"),
            client.todos.create(title="Task 3"),
        ]
        todos = await asyncio.gather(*tasks)

        for todo in todos:
            print(f"Created: {todo.id} - {todo.title}")

        # List all
        result = await client.todos.list()
        print(f"\nTotal: {result.total} todos")

    finally:
        await client.close()

if __name__ == "__main__":
    asyncio.run(main())

Installation and Distribution

Installing Locally

# Using uv (recommended)
cd sdk/python
uv pip install -e .

# Using pip
cd sdk/python
pip install -e .

Publishing to PyPI

cd sdk/python

# Build the package
uv build
# or: python -m build

# Upload to PyPI
uv publish
# or: twine upload dist/*

Installing from Git

pip install git+https://github.com/yourorg/yourrepo.git#subdirectory=sdk/python

​Key Features

​Quick Start

​Step 1: Define Your Contract

​Step 2: Generate the SDK

​Step 3: Install and Use

​Generated Code Structure

​pyproject.toml

​init.py

​Client Configuration

​Creating a Client

​Configuration Options

​Modifying Client Options

​Closing the Client

​Async Client

​Async Context Manager

​Type System

​Type Mapping Reference

​Dataclass Types

​Optional and Nullable Fields

​Enum Fields

​List and Dict Types

​Resources and Methods

​Resource Pattern

​Method Signatures

​Calling Methods

​Streaming (SSE)

​Sync Streaming

​Async Streaming

​Collecting All Events

​Error Handling

​Error Types

​Handling Errors

​Automatic Retries

​Advanced Usage

​Custom Headers

​Authentication Modes

​Timeout Configuration

​Using with Existing httpx Client

​Complete Example

​Server

​Client Usage

​Async Example

​Installation and Distribution

​Installing Locally

​Publishing to PyPI

​Installing from Git

​See Also