klarient 0.3.0

A typed Python framework for modeling and generating API clients.

Klarient

Klarient is a typed Python framework for building small, readable REST API wrappers.

The goal is to let an API wrapper look like the API domain instead of a pile of raw HTTP calls. Resources model the URI tree, request objects model inputs, and response objects model typed results while preserving HTTP metadata.

Install

pip install klarient

Install a transport extra for the HTTP library you want to use:

pip install "klarient[requests]"
pip install "klarient[httpx]"
pip install "klarient[aiohttp]"

Basic Shape

A Klarient wrapper usually has four parts:

• a client class
• resource classes that mirror the REST URI tree
• request objects for typed query or body data
• response objects for typed access to API responses

from enum import StrEnum

from klarient.http.client import _SyncClientImpl
from klarient import (
    JSONBodyRequest,
    RequestField,
    RequestsClient,
    ResourcePath,
    ResponseMap,
    SyncResource,
)


class TaskStatus(StrEnum):
    OPEN = "open"
    DONE = "done"


class TaskCreate(JSONBodyRequest):
    def __init__(
            self,
            *,
            title: str | None = None,
            status: TaskStatus | None = None,
    ) -> None:
        super().__init__()
        self.title = title
        self.status = status

    title = RequestField[str]()
    status = RequestField[TaskStatus]()


class Task(dict[str, object]):
    @property
    def id(self) -> int:
        return int(self["id"])

    @property
    def title(self) -> str:
        return str(self["title"])

    @property
    def status(self) -> TaskStatus:
        return TaskStatus(str(self["status"]))


class TaskResponse(ResponseMap):
    @property
    def data(self) -> Task:
        value = self.get("data", {})
        return Task(value if isinstance(value, dict) else {})


class TaskResource(SyncResource[_SyncClientImpl]):
    def retrieve(self) -> TaskResponse:
        return self._executor.get(TaskResponse)


class TasksResource(SyncResource[_SyncClientImpl]):
    def __getitem__(self, task_id: int | str) -> TaskResource:
        return TaskResource(self, segment=ResourcePath.segment(task_id))

    def create(self, options: TaskCreate) -> TaskResponse:
        return self._executor.post(TaskResponse, options)


class TasksClient(RequestsClient):
    def __init__(self, *, base_url: str) -> None:
        super().__init__(base_url=base_url)
        self.__tasks = self._bind_resource(TasksResource, segment="api/tasks")

    @property
    def tasks(self) -> TasksResource:
        return self.__tasks

Usage:

client = TasksClient(base_url="https://api.example.test")

created = client.tasks.create(
    TaskCreate(title="Write docs", status=TaskStatus.OPEN)
)

print(created.data.id)
print(created.data.title)
print(created.data.status)

task = client.tasks[created.data.id].retrieve()
print(task.data.status)

Request Modeling

Use typed request objects instead of passing dictionaries through your public wrapper API.

from klarient import QueryFieldSpec, QueryRequest, QuerySerialization


class TaskQuery(QueryRequest):
    def __init__(
            self,
            *,
            status: list[TaskStatus] | None = None,
            page: int | None = None,
    ) -> None:
        super().__init__()
        self.status = status
        self.page = page

    status = RequestField[list[TaskStatus]](
        query=QueryFieldSpec(serialization=QuerySerialization.REPEAT)
    )
    page = RequestField[int]()

That repeated field is encoded as repeated query values such as:

?status=open&status=done

Response Modeling

Response models keep the original HTTP response available:

response = client.tasks[123].retrieve()

print(response.data.title)
print(response.status)
print(response.headers)
print(response.native)

Use ResponseMap for JSON objects, ResponseList for JSON arrays, and derive from ResponseBase when a response owns a different representation such as text, XML, or bytes.

Webhook Modeling

Webhook models mirror the response model idea for inbound callbacks. Your web framework receives and routes the HTTP request. Klarient can then normalize that request into a typed object:

class MessageStatusEvent(WebhookForm):
    @property
    def message_sid(self) -> str:
        return str(self["MessageSid"])


event = MessageStatusEvent.from_request(webhook_request)

Use WebhookMap for JSON object payloads, WebhookForm for application/x-www-form-urlencoded callbacks, WebhookList for array payloads, and WebhookText or WebhookBytes for simpler representations. All webhook models keep the source WebhookRequest available through webhook_request.

Project Status

Klarient provides the core framework for building typed REST API wrappers:

• transport-independent HTTP primitives
• sync and async clients
• typed REST resources
• typed request field helpers
• typed response models
• typed webhook models
• reusable pagination strategies
• pluggable transport adapters for requests, httpx, and aiohttp

Examples and broader documentation can live outside the core package so the framework package stays focused on the public runtime API.