Python REST Client

The ThingsBoard Python REST Client wraps the REST API and is auto-generated from the OpenAPI spec — every REST endpoint has a typed method. Edition-specific packages ensure type-safe access to the right endpoints for your tenant.

Installation

Install the edition-specific package:

pip install tb-ce-client==4.3.1.2

Snippets below import models from tb_ce_client.models and exceptions from tb_ce_client.exceptions. Each snippet is self-contained — open it inside a with ThingsboardClient(...) as client: block, replace placeholders like YOUR_API_KEY_VALUE and YOUR_DEVICE_ID with real values, and run it. The client is synchronous and not thread-safe — create one instance per worker.

Quickstart

End-to-end: authenticate with an API key, create a device, push a telemetry value, then clean up.

import json
from tb_ce_client import ThingsboardClient
from tb_ce_client.models import Device

with ThingsboardClient("http://localhost:8080", api_key="YOUR_API_KEY_VALUE") as client:
  saved = client.save_device(Device(name="Quickstart Device", type="default"))
  device_id = str(saved.id.id)

  client.save_entity_telemetry(
      entity_type="DEVICE",
      entity_id=device_id,
      scope="ANY",
      body=json.dumps({"temperature": 22.4}),
  )
  print(f"Pushed telemetry to {saved.name}")

  client.delete_device(device_id=device_id)

Authentication

ThingsboardClient authenticates eagerly on construction. The client is a context manager — using it inside with releases the underlying connection pool on exit. JWT tokens refresh automatically; the client re-logs in transparently if the refresh token expires.

API key (recommended)

from tb_ce_client import ThingsboardClient

with ThingsboardClient("http://localhost:8080", api_key="YOUR_API_KEY_VALUE") as client:
  print(f"Logged in as {client.get_user().email}")

Username and password (JWT)

from tb_ce_client import ThingsboardClient

with ThingsboardClient(
  "http://localhost:8080",
  username="tenant@thingsboard.org",
  password="tenant",
) as client:
  print(f"Logged in as {client.get_user().email}")

Rate-limit handling

The client retries 429 Too Many Requests responses with exponential backoff by default. Tune via constructor keyword arguments (defaults shown below), or pass retry_on_rate_limit=False to disable:

from tb_ce_client import ThingsboardClient

client = ThingsboardClient(
  "http://localhost:8080",
  api_key="YOUR_API_KEY_VALUE",
  max_retries=3,                # default 3
  initial_retry_delay_ms=1_000, # default 1 s
  max_retry_delay_ms=30_000,    # default 30 s
)

Working with entities

Every entity type — devices, assets, dashboards, customers, alarms — follows the same method-naming pattern: save_<entity>, get_<entity>_by_id, get_<entity>_info_by_id, delete_<entity>. Required fields differ per type; see the REST API reference for each model. The walkthrough below uses devices. To update an existing entity, mutate the model and call save_<entity> again.

from tb_ce_client import ThingsboardClient
from tb_ce_client.models import Device

with ThingsboardClient("http://localhost:8080", api_key="YOUR_API_KEY_VALUE") as client:
  new_device = Device(name="Test Device", type="default")
  saved = client.save_device(new_device)
  device_id = str(saved.id.id)  # saved.id is EntityId wrapper, .id is the UUID

  fetched = client.get_device_by_id(device_id=device_id)
  print("Fetched:", fetched.name)

  client.delete_device(device_id=device_id)

Telemetry and attributes

Both save_entity_telemetry(entity_type, entity_id, scope, body) and save_entity_attributes_v2(entity_type, entity_id, scope, body) take a JSON string as body.

Push telemetry

import json
from tb_ce_client import ThingsboardClient

with ThingsboardClient("http://localhost:8080", api_key="YOUR_API_KEY_VALUE") as client:
  device_id = "YOUR_DEVICE_ID"
  client.save_entity_telemetry(
      entity_type="DEVICE",
      entity_id=device_id,
      scope="ANY",  # required by URL path; server-ignored for telemetry
      body=json.dumps({"temperature": 26.5, "humidity": 87}),
  )

Read and write attributes

This example uses an asset with a deviceCount attribute; the same calls work for any entity type — substitute "DEVICE" / device_id for a device.

import json
from tb_ce_client import ThingsboardClient

with ThingsboardClient("http://localhost:8080", api_key="YOUR_API_KEY_VALUE") as client:
  asset_id = "YOUR_ASSET_ID"

  attrs = client.get_attributes_by_scope(
      entity_type="ASSET",
      entity_id=asset_id,
      scope="SERVER_SCOPE",
      keys="deviceCount",  # comma-separated; pass "k1,k2,k3" for multiple
  )

  current = int(attrs[0].value) if attrs else 0
  updated = current + 1

  client.save_entity_attributes_v2(
      entity_type="ASSET",
      entity_id=asset_id,
      scope="SERVER_SCOPE",
      body=json.dumps({"deviceCount": updated}),
  )

Listing and querying entities

Paginated tenant list

from tb_ce_client import ThingsboardClient

with ThingsboardClient("http://localhost:8080", api_key="YOUR_API_KEY_VALUE") as client:
  page = 0  # pages are zero-indexed
  while True:
      devices = client.get_tenant_devices(page_size=100, page=page)
      for device in devices.data:
          print(device.name, str(device.id.id))
      if not devices.has_next:
          break
      page += 1

Filtered query with Entity Data Query API

For server-side filtering on attribute or telemetry values, use the Entity Data Query API. The example below counts all devices, then counts only the active ones:

from tb_ce_client import ThingsboardClient
from tb_ce_client.models import (
  EntityCountQuery, EntityTypeFilter, KeyFilter, EntityKey,
  BooleanFilterPredicate, FilterPredicateValueBoolean,
)

with ThingsboardClient("http://localhost:8080", api_key="YOUR_API_KEY_VALUE") as client:
  type_filter = EntityTypeFilter(entity_type="DEVICE")

  total = client.count_entities_by_query(
      EntityCountQuery(entity_filter=type_filter)
  )
  print("Total devices:", total)

  active_filter = KeyFilter(
      key=EntityKey(type="ATTRIBUTE", key="active"),
      value_type="BOOLEAN",
      predicate=BooleanFilterPredicate(
          operation="EQUAL",
          value=FilterPredicateValueBoolean(default_value=True),
      ),
  )
  active = client.count_entities_by_query(
      EntityCountQuery(entity_filter=type_filter, key_filters=[active_filter])
  )
  print("Active devices:", active)

For a paginated list of matching entities (not just a count), build an EntityDataQuery with an EntityDataPageLink and call client.find_entity_data_by_query(query). See the tb-examples.md catalog for full examples.

Error handling

The client raises ApiException for non-2xx responses, with HTTP-status-specific subclasses (NotFoundException for 404, UnauthorizedException for 401, etc.). e.body is the raw response body (typically JSON for 4xx, may be empty for 5xx).

401 Unauthorized — JWT auth auto-refreshes tokens and re-logs in using stored credentials, so you’ll only see this if credentials become invalid.
429 Too Many Requests — already retried with exponential backoff by default (see Rate-limit handling). You only observe a 429 if retries are exhausted or disabled.
404 Not Found — the common case worth catching explicitly:

from tb_ce_client import ThingsboardClient
from tb_ce_client.exceptions import ApiException, NotFoundException

with ThingsboardClient("http://localhost:8080", api_key="YOUR_API_KEY_VALUE") as client:
  try:
      device = client.get_device_by_id(device_id="nonexistent-id")
  except NotFoundException:
      print("Device not found")
  except ApiException as e:
      print(f"API error {e.status}: {e.body}")

More examples

The tb-examples.md catalog in the repository covers additional entities (assets, dashboards, customers), alarms, relations, and Entity Data Query examples with pagination and sort orders.