klarient 0.2.1

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.

Klarient is early pre-alpha software. The design is usable, but the public API may still change as more real APIs are modeled.

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 import (
    JSONBodyRequest,
    RequestField,
    RequestsTransport,
    ResourcePath,
    ResponseMap,
    SyncClient,
    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__(title=title, 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[SyncClient]):
    def retrieve(self) -> TaskResponse:
        return self._executor.get(TaskResponse)


class TasksResource(SyncResource[SyncClient]):
    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(SyncClient):
    def __init__(self, *, base_url: str) -> None:
        super().__init__(
            base_url=base_url,
            transport=RequestsTransport(),
        )
        self.tasks = TasksResource(self, segment="api/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__(status=status, 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.http_response.headers)
print(response.native_response)

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 is currently focused on the core framework:

• 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 will live outside the core package while the framework stabilizes.