openpygen

openpygen generates typed asynchronous Python clients from OpenAPI 3 specifications. Generated clients use httpx for HTTP requests and Pydantic models for request and response data.

Installation

This project uses Python 3.14 and uv:

uv sync

Generate a client

Run the generator with your own OpenAPI YAML or JSON file:

uv run openpygen generate \
  --input path/to/openapi.yml \
  --output-dir my_service_client \
  --base-url https://api.example.com/v1 \
  --client-name MyServiceClient

--input accepts local YAML and JSON files as well as HTTP(S) URLs. Use --tags payment,billing to generate selected OpenAPI tags only. The output directory name must be a valid Python package name. Add --overwrite to replace an existing output directory.

Generated output

The following snippets illustrate the generated code shape. They use the Payment Service mock specification from the integration test fixtures. The mock API is not part of the public interface of openpygen.

Given an operation such as:

paths:
  /payments:
    post:
      operationId: createPayment
      tags: [payment]
      parameters:
        - name: X-Idempotency-Key
          in: header
          required: false
          schema:
            type: string
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: "#/components/schemas/CreatePaymentRequest"
      responses:
        "201":
          description: Created
          content:
            application/json:
              schema:
                $ref: "#/components/schemas/PaymentResponse"

components:
  schemas:
    CreatePaymentRequest:
      type: object
      required: [amount, currency, merchant_id]
      properties:
        amount:
          type: integer
        currency:
          type: string
        merchant_id:
          type: string

openpygen creates a package with shared transport and exception handling. Each OpenAPI tag receives its own subpackage:

my_service_client/
|-- __init__.py
|-- client.py
|-- exceptions.py
|-- transport.py
|-- payment/
|   |-- __init__.py
|   |-- client.py
|   `-- models.py
`-- billing/
    |-- __init__.py
    |-- client.py
    `-- models.py

Models

Schemas and enums become Pydantic models and StrEnum classes:

from datetime import datetime
from enum import StrEnum

from pydantic import BaseModel, ConfigDict


class PaymentResponseStatus(StrEnum):
    PENDING = "pending"
    COMPLETED = "completed"
    FAILED = "failed"


class CreatePaymentRequest(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    amount: int
    currency: str
    merchant_id: str
    metadata: dict[str, str] | None = None


class PaymentResponse(BaseModel):
    model_config = ConfigDict(populate_by_name=True)

    id: str
    amount: int
    currency: str
    status: PaymentResponseStatus
    created_at: datetime
    metadata: dict[str, str] | None = None

Tag clients

Each tag receives an asynchronous client. Request bodies, headers, query parameters, path parameters, and return values are typed:

from my_service_client.payment.models import (
    CreatePaymentRequest,
    PaymentResponse,
)
from my_service_client.transport import HttpMethod, Transport


class PaymentClient:
    def __init__(self, transport: Transport) -> None:
        self._transport = transport

    async def create_payment(
        self,
        *,
        body: CreatePaymentRequest,
        x_idempotency_key: str | None = None,
    ) -> PaymentResponse:
        headers: dict[str, object] = {}
        if x_idempotency_key is not None:
            headers["X-Idempotency-Key"] = x_idempotency_key
        response = await self._transport.request(
            HttpMethod.POST, "/payments", headers=headers, body=body
        )
        return PaymentResponse.model_validate(response.json())

Aggregate client

The class named by --client-name exposes the generated tag clients:

class MyServiceClient:
    def __init__(
        self,
        *,
        http_client: httpx.AsyncClient | None = None,
        base_url: str | None = None,
    ) -> None:
        self._transport = Transport(http_client=http_client, base_url=base_url)

    @cached_property
    def payment(self) -> PaymentClient:
        return PaymentClient(self._transport)

    @cached_property
    def billing(self) -> BillingClient:
        return BillingClient(self._transport)

Use a generated client

If a base URL was configured during generation, the generated client can be used directly:

import asyncio

from my_service_client import CreatePaymentRequest, MyServiceClient


async def main() -> None:
    async with MyServiceClient() as client:
        payment = await client.payment.create_payment(
            body=CreatePaymentRequest(
                amount=1299,
                currency="EUR",
                merchant_id="merchant_123",
            ),
            x_idempotency_key="order_456",
        )
        print(payment.id, payment.status)


asyncio.run(main())

The base URL can also be set or overridden at runtime:

client = MyServiceClient(base_url="http://localhost:8000")

For tests and custom HTTP configuration, pass an existing httpx.AsyncClient:

import httpx

http_client = httpx.AsyncClient(base_url="http://localhost:8000")
client = MyServiceClient(http_client=http_client)

The caller remains responsible for closing an injected httpx.AsyncClient.

Error handling

Non-successful HTTP responses raise typed exceptions:

from my_service_client import MyServiceClient
from my_service_client.exceptions import NotFoundError


async def load_payment() -> None:
    async with MyServiceClient() as client:
        try:
            payment = await client.payment.get_payment(id="missing")
        except NotFoundError as error:
            print(error.status_code, error.detail)

Generated exception classes cover common status codes, including BadRequestError, UnauthorizedError, NotFoundError, UnprocessableEntityError, RateLimitError, and ServiceUnavailableError.