pyrest-model-client

A simple, flexible Python HTTP client and API modeling toolkit built on top of httpx and pydantic. Easily integrate robust API requests and resource models into your Python projects.

🚀 Features

• Model-driven: Define and interact with API resources as Python classes using BaseAPIModel.
• Easy HTTP Requests: RestApiClient for GET, POST, PUT, PATCH, DELETE with automatic header and base URL management.
• Async Support: Full async/await support with AsyncRestApiClient for high-performance concurrent requests.
• Automatic Endpoint Normalization: Configurable endpoint path normalization (trailing slash handling).
• Resource Path Integration: Models can use their resource_path to generate endpoints and URLs automatically.
• Flexible Authentication: Support for Token and Bearer authentication via build_header() helper.
• Response to Model Conversion: get_model_fields() helper converts API responses to typed model instances.
• Configurable Client: Customizable timeout, connection pool limits, and redirect handling.
• Type Safety: All models use Pydantic for automatic validation and serialization.
• Error Handling: Automatic HTTP status error handling with raise_for_status().
• Extensible: Easily create new models for any RESTful resource by extending BaseAPIModel.

📦 Installation

uv add pyrest-model-client

🔧 Usage

1. Define Your Models

from pyrest_model_client.base import BaseAPIModel


class User(BaseAPIModel):
    name: str
    email: str
    resource_path: str = "user"


class Environment(BaseAPIModel):
    name: str
    resource_path: str = "environment"

2. Initialize the Client and Make Requests

import os

from dotenv import load_dotenv

from pyrest_model_client import BaseAPIModel, RestApiClient, build_header, get_model_fields

load_dotenv()

TOKEN = os.getenv("TOKEN")
BASE_URL = f'{os.getenv("BASE_URL")}:{os.getenv("PORT")}'


class FirstApp(BaseAPIModel):
    name: str
    description: str | None = None
    resource_path: str = "first_app"


header = build_header(token=TOKEN)

# Use as a context manager (auto-closes the connection)
# Optionally configure timeout and connection pool limits:
#   timeout=httpx.Timeout(60.0, connect=10.0)
#   limits=httpx.Limits(max_keepalive_connections=5, max_connections=10)
with RestApiClient(base_url=BASE_URL, header=header) as client:
    # Example: Use resource_path from model
    app = FirstApp(name="My App", description="Test")
    endpoint = app.get_endpoint()           # Returns "first_app"
    full_url = app.get_resource_url(client) # Returns full URL

    # Example: Get all items (paginated) — get_model_fields returns list[FirstApp],
    # so type checkers infer the concrete subclass on every element.
    item_list: list[FirstApp] = []
    params = None
    while True:
        res = client.get("first_app", params=params)
        data = res.json()
        item_list.extend(get_model_fields(data["results"], model=FirstApp))
        if not data["next"]:
            break
        params = {"page": data["next"].split("/?page=")[-1]}

    # Example: Create a new item
    new_item = client.post("first_app", data={"name": "My App", "description": "A new app"})

    # Example: Full update
    updated_item = client.put("first_app/1", data={"name": "Updated App"})

    # Example: Partial update
    patched_item = client.patch("first_app/1", data={"description": "New description"})

    # Example: Delete an item
    client.delete("first_app/1")

3. Using Async Client

import asyncio
import os

from dotenv import load_dotenv

from pyrest_model_client import AsyncRestApiClient, build_header

load_dotenv()

TOKEN = os.getenv("TOKEN")
BASE_URL = f'{os.getenv("BASE_URL")}:{os.getenv("PORT")}'


async def main() -> None:
    header = build_header(token=TOKEN)

    async with AsyncRestApiClient(base_url=BASE_URL, header=header) as client:
        data = (await client.get("first_app")).json()
        await client.post("first_app", data={"name": "Async App"})
        await client.put("first_app/1", data={"name": "Updated"})
        await client.patch("first_app/1", data={"description": "Patched"})
        await client.delete("first_app/1")


asyncio.run(main())

🤝 Contributing

Contributions are welcome! Please fork the repo, create a branch, and submit a pull request.

📄 License

MIT License — see LICENSE for details.