koios-client 1.0.0

Python client library for the Koios IoT platform — typed GraphQL & REST wrapper

Koios Client

Typed Python SDK for the Koios IoT platform. Provides two API layers:

• High-level resource API — Device, Tag, Model objects with intuitive methods like .enable(), .update(), .delete()
• Low-level GraphQL client — fully-typed generated client via client.gql with IDE autocomplete

Plus REST helpers for file operations, import/export, backups, and trend data.

Installation

pip install koios-client

Or install from source:

pip install -e ".[dev]"

Quick Start

from koios_client import KoiosClient

client = KoiosClient(
    hostname="koios.example.com",
    client_id="your-client-uuid",
    client_secret="your-client-secret",
)

# High-level resource API — fetch by ID (default) or slug
device = client.device(1)
device.enable()
for tag in device.tags():
    print(tag.name, tag.value)

# Low-level GraphQL (fully typed — IDE autocomplete works)
result = client.gql.get_devices()
for device in result.devices.results:
    print(device.name, device.status)

client.close()

Use as a context manager for automatic cleanup:

with KoiosClient("koios.example.com", "client-id", "secret") as client:
    device = client.device(1)
    print(device.name, device.enabled)

High-Level Resource API

The resource layer wraps the generated GraphQL client with intuitive Python objects. All resource classes are importable from the top-level package:

from koios_client import Device, Tag, Model, PaginatedList

Fetching Resources

Every resource can be fetched by ID (default positional arg) or slug (UUID).

# By ID (default — integer database primary key)
device = client.device(1)
tag = client.tag(42)
model = client.model(5)

# By slug (UUID — useful when you already have the identifier)
device = client.device(slug="550e8400-e29b-41d4-a716-446655440000")

# By name — use the list query with filters
from koios_client.types import DeviceFilter, StrFilterLookup
devices = client.devices(filters=DeviceFilter(name=StrFilterLookup(exact="My Device")))
device = devices[0]

# List with filtering, ordering, and pagination
from koios_client.types import OffsetPaginationInput

devices = client.devices(
    filters=DeviceFilter(enabled=True),
    pagination=OffsetPaginationInput(limit=50),
)
print(f"{devices.total_count} total devices")
for device in devices:
    print(device.name, device.slug, device.enabled)

PaginatedList supports iteration, indexing, len(), and bool():

tags = client.tags()
print(len(tags))       # Number of items on this page
print(tags.total_count) # Total matching items on server
first = tags[0]         # Index access

CRUD Operations

from koios_client.types import DeviceInput, OneToManyInput

# Create — `protocol` takes a OneToManyInput pointing at the protocol's ID
device = client.create_device(DeviceInput(
    name="New Device",
    protocol=OneToManyInput(set="1"),
    scan_rate=5.0,
))

# Update (returns a fresh object with all fields refreshed)
device = device.update(name="Renamed Device", scan_rate=5.0, enabled=True)

# Enable / Disable (convenience wrappers around update)
device = device.enable()
device = device.disable()

# Duplicate
copy = device.duplicate("Device Copy", description="Cloned from original")

# Delete
device.delete()

Parent-Child Accessors

# Device → Tags
device = client.device(1)
tags = device.tags(
    filters=TagFilter(enabled=True),
    pagination=OffsetPaginationInput(limit=100),
)

# Model → Bindings
model = client.model(5)
for binding in model.bindings():
    print(binding.name, binding.usage, binding.normalization_type)
    binding.update(normalization_type="MIN_MAX")

Bulk Operations

# Bulk enable / disable
client.enable_devices(["1", "2", "3"])
client.disable_tags(["10", "11", "12"])

# Bulk delete
client.delete_devices(["1", "2", "3"])
client.delete_tags(["10", "11", "12"])
client.delete_models(["20", "21"])
client.delete_scan_groups(["30", "31"])
client.delete_device_sets(["40", "41"])

Live Data

Fetch real-time values from the Redis cache:

device = client.device(1)
live = device.live()
print(live.get("status"), live.get("error_message"))

tag = client.tag(42)
live = tag.live()
print(live.get("value"), live.get("timestamp"), live.get("quality"))

Import / Export with Resources

# Import with dry_run (default) — returns preview without applying
preview = client.import_devices("devices.csv")
print(preview)

# Import for real
result = client.import_devices("devices.csv", dry_run=False)

# Export a single device/model
csv_bytes = device.export_csv()
zip_bytes = model.export()

All Resource Types

All singular methods accept id (positional, integer) or slug= (UUID). Exactly one must be provided.

Method	Returns	Notes
client.device(id, slug=)	Device
client.devices(...)	PaginatedList[Device]	filters, order, pagination
client.create_device(data)	Device
client.enable_devices(ids)	list[Device]
client.disable_devices(ids)	list[Device]
client.delete_devices(ids)	None
client.import_devices(file)	dict	dry_run=True by default
client.tag(id, slug=)	Tag
client.tags(...)	PaginatedList[Tag]
client.create_tag(data)	Tag
client.enable_tags(ids)	list[Tag]
client.disable_tags(ids)	list[Tag]
client.delete_tags(ids)	None
client.import_tags(file)	dict	dry_run=True by default
client.model(id, slug=)	Model
client.models(...)	PaginatedList[Model]
client.create_model(data)	Model
client.enable_models(ids)	list[Model]
client.disable_models(ids)	list[Model]
client.delete_models(ids)	None
client.import_models(file)	dict	dry_run=True by default
client.scan_group(id, slug=)	ScanGroup
client.scan_groups(...)	PaginatedList[ScanGroup]
client.create_scan_group(data)	ScanGroup
client.enable_scan_groups(ids)	list[ScanGroup]
client.disable_scan_groups(ids)	list[ScanGroup]
client.delete_scan_groups(ids)	None
client.device_set(id, slug=)	DeviceSet	supports .update(), .delete(), .set_active_device(item_id)
client.device_sets(...)	PaginatedList[DeviceSet]
client.create_device_set(data)	DeviceSet
client.delete_device_sets(ids)	None
client.protocol(id, slug=)	Protocol	read-only, supports .set_visible(bool)
client.protocols()	list[Protocol]	not paginated

Device Sets (Redundancy Groups)

Device sets group multiple devices for failover. Member devices are added as DeviceSetItem records via the low-level GraphQL client, and the active member is controlled by set_active_device:

from koios_client.types import (
    DeviceSetInput,
    DeviceSetItemInput,
    OneToManyInput,
)

ds = client.create_device_set(DeviceSetInput(
    name="Redundant Pair",
    protocol=OneToManyInput(set="1"),
))

# Add two member devices with priority 1 (primary) and 2 (backup).
item1 = KoiosClient.check_operation(
    client.gql.create_device_set_item(
        data=DeviceSetItemInput(
            device_set=OneToManyInput(set=ds.id),
            device=OneToManyInput(set="10"),
            priority=1,
        )
    ).create_device_set_item
)
item2 = KoiosClient.check_operation(
    client.gql.create_device_set_item(
        data=DeviceSetItemInput(
            device_set=OneToManyInput(set=ds.id),
            device=OneToManyInput(set="11"),
            priority=2,
        )
    ).create_device_set_item
)

# Switch which member is active — identified by the DeviceSetItem ID, not the Device ID.
ds = ds.set_active_device(item2.id)

Endpoint Clients

# List endpoint clients
clients = client.endpoint_clients()

# Get a single endpoint client by ID
ec = client.endpoint_client("123")

# Create a new endpoint client (returns one-time secret)
result = client.create_endpoint_client("My Integration", description="Data pipeline")

# Set permissions
client.set_endpoint_client_permissions(
    client_id="123",
    permission_ids=["1", "2", "3"],
)

Low-Level GraphQL Client

All GraphQL queries and mutations are available via client.gql with fully typed parameters and return values.

Querying Data

from koios_client.types import (
    DeviceFilter,
    OffsetPaginationInput,
    StatusChoices,
)

# Paginated queries
devices = client.gql.get_devices(
    pagination=OffsetPaginationInput(limit=50),
)
print(f"Total: {devices.devices.total_count}")
for device in devices.devices.results:
    print(device.name, device.status, device.protocol.name)

# Filtered queries
tags = client.gql.get_tags(
    filters=TagFilter(enabled=True),
    pagination=OffsetPaginationInput(limit=100),
)

# Single item by slug
device = client.gql.get_device(slug="my-device")
tag = client.gql.get_tag(slug="my-tag")

Mutations

Single-item mutations return a union of the entity type or OperationInfo. Use check_operation to unwrap:

from koios_client.types import DeviceInput, OneToManyInput

raw = client.gql.create_device(data=DeviceInput(
    name="New Device",
    protocol=OneToManyInput(set="1"),
))

# Raises OperationError if the server returned validation/permission errors
device = KoiosClient.check_operation(raw.create_device)
print(f"Created: {device.name} ({device.slug})")

Available Enums

All GraphQL enums are available as typed Python enums:

from koios_client.types import (
    StatusChoices,        # RUNNING, STOPPED, FAILED
    UsageChoices,         # INPUT, OUTPUT, VIRTUAL
    AggregateFunction,    # MEAN, SUM, MIN, MAX, FIRST, LAST, COUNT
    DeviceErrorCodeChoices,
    TagErrorCodeChoices,
    ProtocolReferenceCodeChoices,  # OPCUA, MODBUS_TCP, ETHERNET_IP, ...
)

REST API

Operations that involve file uploads, background tasks, or binary responses use REST endpoints exposed directly on the client.

Export / Import

# Export devices as CSV
csv_bytes = client.export_devices_csv()
with open("devices.csv", "wb") as f:
    f.write(csv_bytes)

# Export specific tags
csv_bytes = client.export_tags_csv(ids=[1, 2, 3])

# Export models as ZIP (includes bindings)
zip_bytes = client.export_models()

# Two-step import: preview then confirm
preview = client.import_tags_preview("tags.csv")
print(f"Will import {preview['result']['total_rows']} rows")
if preview["can_import"]:
    result = client.import_tags_confirm(
        preview["tmp_storage_name"],
        preview["file_name"],
    )
    print(f"Imported {result['imported_count']} tags")

Model Files

# Upload a model file
result = client.upload_model_file(
    "model.onnx",
    model_slug="my-model",
    version="1.0",
    set_active=True,
)

# Download a model file
model_bytes = client.download_model_file("model-file-slug")

# Get model structure (weights stripped for ONNX)
structure = client.get_model_file_structure("model-file-slug")

Backups

# Create a backup
task = client.create_backup(tier="full")

# Poll for completion
import time
while True:
    status = client.get_backup_status(task["task_id"])
    if status["status"] == "completed":
        break
    time.sleep(2)

# List and download
backups = client.list_backups()
for backup in backups["backups"]:
    print(backup["filename"], backup["size_bytes"])

data = client.download_backup(backups["backups"][0]["filename"])

# Restore from backup
upload = client.upload_restore_file("backup.tar.gz")
task = client.start_restore(upload["restore_file_path"])

Trend Data Export

# Start an export task
task = client.create_trend_export(
    tag_ids=[1, 2, 3],
    start="2026-01-01T00:00:00Z",
    stop="2026-02-01T00:00:00Z",
    mode="resampled",
    resample_interval="5m",
    aggregate_fn="mean",
    output_format="csv",
)

# Poll and download
status = client.get_trend_export_status(task["task_id"])
if status["status"] == "completed":
    data = client.download_trend_export(status["filename"])

# Estimate before exporting
estimate = client.estimate_trend_export(
    tag_ids=[1, 2, 3],
    export_all=True,
)
print(f"~{estimate['estimated_size_bytes'] / 1e6:.1f} MB")

Component Libraries

result = client.upload_component_library("my-component.kcl")
print(f"Uploaded: {result['name']} v{result['version']}")

OPC-UA Certificates

# Upload a certificate + key pair
result = client.upload_opcua_certificate(
    "cert.der", "key.pem", name="My Certificate"
)

# Download cert or key
cert_bytes = client.download_opcua_certificate("cert-slug")
key_bytes = client.download_opcua_key("cert-slug")

EDS Files

# Upload an EDS file for a device
result = client.upload_eds_file("device.eds")

# Delete the EDS file from a device
client.delete_eds_file(device_id=1)

Log Downloads

log = client.download_device_log("my-device")
log = client.download_model_log("my-model")
log = client.download_scan_group_log("scan-group-1")
log = client.download_component_log("instance-1")
log = client.download_service_log("datacollector")

Connection Options

client = KoiosClient(
    hostname="koios.example.com",
    client_id="your-uuid",
    client_secret="your-secret",
    port=443,           # Default: 443 (HTTPS) or 80 (HTTP)
    ssl=True,           # Default: True
    verify_ssl=True,    # Default: True (set False for self-signed certs)
    timeout=30.0,       # Default: 30 seconds
)

Error Handling

from koios_client import (
    KoiosError,           # Base exception
    AuthenticationError,  # Bad credentials or expired token
    NotFoundError,        # Resource not found (e.g. client.device(999))
    OperationError,       # Mutation returned OperationInfo
    GraphQLError,         # GraphQL response errors
    KoiosConnectionError, # Network connectivity issues
    KoiosPermissionError, # Insufficient permissions
    ValidationError,      # Input validation failures
)

try:
    device = client.device(1)
    device.update(name="New Name")
except NotFoundError:
    print("Device not found")
except OperationError as e:
    for msg in e.messages:
        print(f"{msg.kind}: {msg.field}: {msg.message}")
except AuthenticationError:
    print("Check your client credentials")

Development

# Install with dev dependencies
make install

# Format and lint
make format      # ruff check --fix + format
make lint        # ruff check + format --check

# Type check and run unit tests
make typecheck   # basedpyright
make test        # pytest (mocked HTTP; no live server needed)
make ci          # lint + typecheck + test

# Regenerate GraphQL client (requires a running Koios dev server at :8080)
make codegen

Integration tests

make test-integration runs the suite under tests/integration/ against a live Koios server. These tests are skipped by the default make test / make ci runs — they exist so releases can be verified against a real container.

Copy tests/integration/.env.example to tests/integration/.env and fill in an endpoint client ID/secret from the target server:

cp tests/integration/.env.example tests/integration/.env
# Edit tests/integration/.env with KOIOS_HOST, KOIOS_CLIENT_ID, KOIOS_CLIENT_SECRET
make test-integration

Releasing

# Bump pyproject.toml version, land via PR, then tag:
make tag         # reads version from pyproject.toml, creates matching git tag

License

Copyright 2024-2026 Ai-OPs, Inc. All rights reserved.