Metadata-Version: 2.4
Name: kelvin-python-api-client
Version: 1.1.4b1
Summary: Kelvin Python API Client
Author-email: Kelvin Inc <engineering@kelvininc.com>
Project-URL: Homepage, https://kelvin.ai/
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Requires-Dist: PyYAML==6.*
Requires-Dist: pydantic-settings==2.*
Requires-Dist: pydantic[email]==2.13.*
Requires-Dist: python-keycloak==5.*
Requires-Dist: httpx[http2]==0.28.*
Requires-Dist: structlog==23.*
Requires-Dist: typing-extensions==4.*
Requires-Dist: typing-inspect==0.9.*
Requires-Dist: tzlocal==5.2.*
Requires-Dist: kelvin-krn<0.2,>=0.1.3b1
Provides-Extra: dataframe
Requires-Dist: pandas==2.*; extra == "dataframe"

# 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.

```python
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](https://www.kelvininc.com/license-sdk) for more information.

## Table of Contents

- [Installation](#installation)
- [Authentication](#authentication)
  - [Username & password](#username--password)
  - [Client ID & secret (service accounts)](#client-id--secret-service-accounts)
  - [Pre-fetched access token](#pre-fetched-access-token)
  - [Environment variables](#environment-variables)
  - [Other client options](#other-client-options)
- [Sync and Async](#sync-and-async)
- [Discovering the API](#discovering-the-api)
  - [What resources and methods exist](#what-resources-and-methods-exist)
  - [Inspecting request and response models](#inspecting-request-and-response-models)
- [Passing request bodies](#passing-request-bodies)
- [Controlling what a method returns](#controlling-what-a-method-returns)
  - [`dry_run` — inspect the request without sending it](#dry_run--inspect-the-request-without-sending-it)
  - [`_get_response` — get the raw HTTP response](#_get_response--get-the-raw-http-response)
- [Pagination](#pagination)
  - [Auto-fetch everything (default)](#auto-fetch-everything-default)
  - [One page at a time](#one-page-at-a-time)
  - [Streaming iterators](#streaming-iterators)
- [Working with results](#working-with-results)
- [Error handling](#error-handling)
- [Logging requests and responses](#logging-requests-and-responses)

## Installation

```bash
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:

```bash
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](#environment-variables).

### Username & password

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

```python
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:

```python
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:

```python
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.

```python
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:

| Variable | Constructor arg |
|----------|-----------------|
| `KELVIN_CLIENT__URL` | `url` |
| `KELVIN_CLIENT__USERNAME` | `username` |
| `KELVIN_CLIENT__PASSWORD` | `password` |
| `KELVIN_CLIENT__TOTP` | `totp` |
| `KELVIN_CLIENT__CLIENT_ID` | `client_id` |
| `KELVIN_CLIENT__CLIENT_SECRET` | `client_secret` |
| `KELVIN_CLIENT__RETRIES` | `retries` |
| `KELVIN_CLIENT__TIMEOUT` | `timeout` |

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

```python
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

| Argument | Default | Description |
|----------|---------|-------------|
| `retries` | `3` | Number of automatic retries on transient failures |
| `timeout` | `(6, 10)` | `read` timeout, or `(connect, read)` tuple, in seconds |
| `verbose` | `False` | Log every request and response (see [Logging](#logging-requests-and-responses)) |

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

```python
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`.

```python
# Synchronous
from kelvin.api.client import Client

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

```python
# 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:

```python
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/`](src/kelvin/api/client/api/) (sync) and
[`src/kelvin/api/client/async_api/`](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/`](src/kelvin/api/client/model/):

| Module | Contains |
|--------|----------|
| `requests` | Request body models (what you send) |
| `responses` | Top-level response models, including the paginated wrappers |
| `type` | The element/entity models (e.g. `Asset`, `Datastream`) |
| `enum` | Enumerations used by the models |
| `pagination` | `PaginationCursor` / `PaginationLimits` page-info models |

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

```python
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:

```python
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:

```python
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.

```python
# 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.

```python
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:

```python
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`:

```python
# 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)
```

| `pagination_type` | Page-info fields (`page.pagination`) | How to get the next page |
|-------------------|--------------------------------------|--------------------------|
| `cursor` (default) | `next_page`, `previous_page` | pass the bookmark as `next=` / `previous=` |
| `limits` | `page`, `page_size`, `total_pages`, `total_items` | increment `page=` |

### 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:

```python
# 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):

```python
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:

```python
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`](https://www.structlog.org/):

```python
import logging

logging.basicConfig(level=logging.DEBUG)

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