kelvin-python-api-client

A Python client for the Kelvin platform REST API. It gives you typed, synchronous and asynchronous access to every Kelvin resource — assets, datastreams, time series, recommendations, control changes, apps, workloads, users, and more.

The client handles authentication, token refresh, retries, pagination, and request/response validation for you, so you can focus on the data.

from kelvin.api.client import Client

client = Client()  # reads credentials from env vars
assets = client.asset.list_assets()  # returns a typed, fully-paginated list

for asset in assets:
    print(asset.name)

df = assets.to_df()  # or work with a pandas DataFrame

License

See the License for more information.

Table of Contents

• Installation
• Authentication
  • Username & password
  • Client ID & secret (service accounts)
  • Pre-fetched access token
  • Environment variables
  • Other client options
• Sync and Async
• Discovering the API
  • What resources and methods exist
  • Inspecting request and response models
• Passing request bodies
• Controlling what a method returns
  • dry_run — inspect the request without sending it
  • _get_response — get the raw HTTP response
• Pagination
  • Auto-fetch everything (default)
  • One page at a time
  • Streaming iterators
• Working with results
• Error handling
• Logging requests and responses

Installation

pip install kelvin-python-api-client

The client can convert results to pandas DataFrames. That feature is optional — install the extra if you want it:

pip install kelvin-python-api-client[dataframe]

Authentication

The client authenticates against Kelvin's Keycloak instance and manages the access/refresh token lifecycle automatically — tokens are fetched on the first request and refreshed before they expire. You just choose how to provide credentials.

All authentication parameters are accepted directly by Client(...) (and AsyncClient(...)). Anything you don't pass falls back to the environment variables.

Username & password

The most common interactive flow. Add totp if the account has 2FA enabled.

from kelvin.api.client import Client

client = Client(
    url="https://my-instance.kelvininc.com",
    username="me@example.com",
    password="••••••••",
    totp=123456,  # optional, only if 2FA is enabled
)

You can also defer the credentials and log in later:

client = Client(url="https://my-instance.kelvininc.com", username="me@example.com")
client.login(password="••••••••", totp=123456)

Client ID & secret (service accounts)

For non-interactive / automation use, authenticate with a service account using the OAuth2 client credentials grant:

client = Client(
    url="https://my-instance.kelvininc.com",
    client_id="my-service-account",
    client_secret="••••••••",
)

If you don't set client_id, it defaults to kelvin-client (the public client used for the username/password flow).

Pre-fetched access token

If you already have a valid bearer token (e.g. injected by the runtime your code runs in), pass it directly. The client will use it as-is and won't attempt to log in or refresh.

client = Client(
    url="https://my-instance.kelvininc.com",
    access_token="eyJhbGciOiJ...",
)

Environment variables

Any constructor argument can be supplied via environment variables instead. They are nested under the KELVIN_CLIENT prefix with a __ (double underscore) delimiter:

export KELVIN_CLIENT__URL="https://my-instance.kelvininc.com"
export KELVIN_CLIENT__USERNAME="me@example.com"
export KELVIN_CLIENT__PASSWORD="••••••••"

from kelvin.api.client import Client

client = Client()  # everything pulled from the environment

Explicit constructor arguments always take precedence over environment variables.

Other client options

When you're done, close the client to release connections. Both clients are also context managers, which is the recommended pattern:

with Client() as client:
    assets = client.asset.list_assets()
# connection pool closed automatically

Sync and Async

Every resource and method exists in two flavours. They are identical in name and signature — the only difference is await.

# Synchronous
from kelvin.api.client import Client

client = Client()
assets = client.asset.list_assets()

# Asynchronous
import asyncio
from kelvin.api.client import AsyncClient


async def main():
    async with AsyncClient() as client:
        assets = await client.asset.list_assets()


asyncio.run(main())

Use the async client when you want concurrency (e.g. firing many requests with asyncio.gather) or when integrating into an async application. Everything below applies to both; just add await for the async client.

Discovering the API

The set of resources and methods follows the Kelvin platform API and changes over time as the platform evolves. Rather than memorising a list, learn how to discover what's available from the code itself.

What resources and methods exist

Resources hang off the client as attributes (client.asset, client.datastreams, client.timeseries, …). To see them and their methods, use dir() or your IDE's autocomplete:

from kelvin.api.client import Client

client = Client()

dir(client)  # -> all resource names: 'asset', 'datastreams', 'timeseries', ...
dir(client.asset)  # -> all methods on the asset resource
help(client.asset.list_assets)  # -> full docstring, args, endpoint, required permission

In the repository itself, the resources live under src/kelvin/api/client/api/ (sync) and src/kelvin/api/client/async_api/ (async). Each file is one resource (asset.py, timeseries.py, …) and each public method maps to one REST endpoint. Every method's docstring states the HTTP verb, path, and the permission it requires, for example:

``listAssets``: ``GET`` ``/api/v4/assets/list``
**Permission Required:** `kelvin.permission.asset.read`.

Inspecting request and response models

The request and response shapes are Pydantic models, generated from the API spec, in src/kelvin/api/client/model/:

Because they're Pydantic models, you can introspect them at runtime:

from kelvin.api.client.model import requests

requests.TimeseriesRangeGet.model_fields  # field names, types, defaults
print(requests.TimeseriesRangeGet.model_json_schema())  # full JSON schema
help(requests.TimeseriesRangeGet)  # docstring with the field list

Passing request bodies

Methods that send a body accept the data argument in two interchangeable forms.

1. As a typed model — gives you validation and autocomplete:

from datetime import datetime, timedelta
from kelvin.api.client import Client
from kelvin.api.client.model import requests

client = Client()

data = client.timeseries.get_timeseries_range(
    data=requests.TimeseriesRangeGet(
        selectors=[{"resource": "krn:ad:my-asset/temperature"}],
        start_time=datetime.now() - timedelta(hours=1),
        end_time=datetime.now(),
    )
)

2. As a plain dict — convenient for quick scripts; it's validated against the same model under the hood:

data = client.timeseries.get_timeseries_range(
    data={
        "selectors": [{"resource": "krn:ad:my-asset/temperature"}],
        "start_time": "2026-01-01T00:00:00Z",
        "end_time": "2026-01-01T01:00:00Z",
    }
)

You may also pass the individual fields as keyword arguments instead of a data object — they're collected into the request model for you.

Controlling what a method returns

Most methods return a parsed, typed result by default. Two special flags change that behaviour for debugging and advanced use.

dry_run — inspect the request without sending it

Note: there are two different things named "dry run", don't confuse them:

• dry_run (no leading underscore) is a server-side feature on some write endpoints. It is sent to the API, which validates the operation and returns feedback without persisting any changes. The request is sent.
• _dry_run=True is a client-side flag. The request is not sent at all; instead the method returns the request it would have made as a dict. Useful for inspecting the exact path, params, and body.

# Server-side dry run: validates the create, changes nothing
client.asset.create_asset_bulk(dry_run=True, data=my_payload)

# Client-side: don't send anything, just show me the request
req = client.asset.create_asset_bulk(_dry_run=True, data=my_payload)
print(req)
# {'method': 'POST', 'path': '/api/v4/assets/bulk/create', 'data': {...}, 'params': {...}, ...}

_get_response — get the raw HTTP response

Pass _get_response=True to receive the underlying httpx.Response instead of a parsed model. The client does not raise on error statuses in this mode, so you inspect the status and body yourself.

resp = client.asset.list_assets(_get_response=True)
print(resp.status_code)
print(resp.headers)
print(resp.json())

This is also the only mode that supports pagination_type="stream" on list endpoints (see below).

Pagination

List endpoints are paginated. The client exposes three styles, selected with the pagination_type argument: cursor (default), limits (page numbers), and stream.

Auto-fetch everything (default)

By default (fetch=True), list methods transparently follow every page and return a single KList containing all items — you never deal with cursors or page numbers:

assets = client.asset.list_assets()  # all assets, every page already fetched
print(len(assets))
for asset in assets:
    print(asset.name)

This works for both cursor and limits pagination.

One page at a time

Pass fetch=False to get back the raw paginated response for a single page, including the pagination metadata so you can walk pages yourself. The concrete type depends on pagination_type:

# Cursor pagination: page info carries next/previous bookmarks
page = client.asset.list_assets(fetch=False, pagination_type="cursor", page_size=100)
for asset in page.data:
    print(asset.name)
next_bookmark = page.pagination.next_page  # pass as `next=` to get the next page
if next_bookmark:
    page2 = client.asset.list_assets(fetch=False, next=next_bookmark, page_size=100)

# Limits pagination: page info carries page numbers and totals
page = client.asset.list_assets(fetch=False, pagination_type="limits", page=1, page_size=100)
print(page.pagination.page, "of", page.pagination.total_pages)
print(page.pagination.total_items)

Streaming iterators

Some endpoints (notably time-series reads like get_timeseries_range) return a lazy iterator instead of a list. Iterate it directly — data is consumed as it arrives over the wire, which keeps memory flat for large result sets:

# Sync: a KIterator — iterate with a normal for-loop
data = client.timeseries.get_timeseries_range(data={...})
for point in data:
    print(point)

# Async: an AsyncKIterator — iterate with async for
data = await aclient.timeseries.get_timeseries_range(data={...})
async for point in data:
    print(point)

pagination_type="stream" on a regular list endpoint asks the server to return all results in one streamed response. It is only available together with _get_response=True (raw response); if you request it without that flag the client silently falls back to cursor pagination.

Working with results

KList is a normal Python list subclass, so it indexes, slices, and iterates like any list. Both KList and the streaming iterators add a to_df() helper that returns a pandas DataFrame (requires the [dataframe] extra):

assets = client.asset.list_assets()
df = assets.to_df()  # one row per asset, columns flattened

# Time-series iterators support long (default) or wide format:
data = client.timeseries.get_timeseries_range(data={...})
df = data.to_df()  # long: timestamp | asset_name | datastream_name | payload
df = data.to_df(datastreams_as_column=True)  # wide: one column per datastream

Individual items are Pydantic models — use .model_dump() for a dict or .model_dump_json() for JSON.

Error handling

When a request fails (and you didn't pass _get_response=True), the client raises an exception. All client errors derive from ClientError, so you can catch broadly or specifically:

ClientError                     # base of everything below
├── AuthenticationError
│   ├── LoginError              # wrong credentials, missing auth params
│   └── LogoutError
├── APIError                    # API returned a 4xx/5xx with a structured error body
└── ResponseError               # unexpected/unparseable response

The most common one to handle is APIError. It carries the originating httpx.Response and a parsed list of error objects:

from kelvin.api.client import Client
from kelvin.api.base.error import APIError

client = Client()

try:
    client.asset.get_asset(asset_name="does-not-exist")
except APIError as exc:
    print(exc.response.status_code)  # e.g. 404
    for err in exc.errors:  # parsed error objects from the response body
        print(err.title, "-", err.description)
    print(exc)  # full summary: method, url, status, error body

APIError, LoginError, and the rest all live in kelvin.api.base.error (ClientError is also re-exported from kelvin.api.client).

Logging requests and responses

To see exactly what goes over the wire while debugging, construct the client with verbose=True. Requests and responses (URL, headers, body) are emitted at DEBUG level via structlog:

import logging

logging.basicConfig(level=logging.DEBUG)

client = Client(verbose=True)
client.asset.list_assets()