Python REST Client (Legacy)

The ThingsBoard Python REST Client wraps the REST API and lets Python scripts manage devices, assets, dashboards, customers, and other platform entities. It supports context managers for automatic token refresh.

Installation

pip3 install tb-rest-client==4.3.1.6

Authentication

API key (recommended)

import logging
from tb_rest_client.rest_client_ce import *
from tb_rest_client.rest import ApiException

logging.basicConfig(level=logging.DEBUG,
                  format='%(asctime)s - %(levelname)s - %(module)s - %(lineno)d - %(message)s',
                  datefmt='%Y-%m-%d %H:%M:%S')

url = "https://eu.thingsboard.cloud"
api_key = "YOUR_API_KEY_HERE"

def main():
  with RestClientCE(base_url=url) as rest_client:
      try:
          rest_client.api_key_login(api_key)
          device_profile = DeviceProfile(name="Thermometer",
                                         type="DEFAULT",
                                         transport_type="DEFAULT",
                                         profile_data=DeviceProfileData(
                                             configuration={"type": "DEFAULT"},
                                             transport_configuration={"type": "DEFAULT"}))
          device_profile = rest_client.save_device_profile(device_profile)
          device = Device(name="Thermometer 1", label="Thermometer 1",
                          device_profile_id=device_profile.id)
          device = rest_client.save_device(device)
          logging.info("Device was created:\n%r\n", device)
      except ApiException as e:
          logging.exception(e)

if __name__ == '__main__':
  main()

Username and password (deprecated)

import logging
from tb_rest_client.rest_client_ce import *
from tb_rest_client.rest import ApiException

logging.basicConfig(level=logging.DEBUG,
                  format='%(asctime)s - %(levelname)s - %(module)s - %(lineno)d - %(message)s',
                  datefmt='%Y-%m-%d %H:%M:%S')

url = "https://eu.thingsboard.cloud"
username = "[email protected]"
password = "tenant"

def main():
  with RestClientCE(base_url=url) as rest_client:
      try:
          rest_client.login(username=username, password=password)
          default_asset_profile_id = rest_client.get_default_asset_profile_info().id
          asset = Asset(name="Building 12", asset_profile_id=default_asset_profile_id)
          asset = rest_client.save_asset(asset)
          logging.info("Asset was created:\n%r\n", asset)

          device_profile = DeviceProfile(name="Thermometer",
                                         type="DEFAULT",
                                         transport_type="DEFAULT",
                                         profile_data=DeviceProfileData(
                                             configuration={"type": "DEFAULT"},
                                             transport_configuration={"type": "DEFAULT"}))
          device_profile = rest_client.save_device_profile(device_profile)
          device = Device(name="Thermometer 1", device_profile_id=device_profile.id)
          device = rest_client.save_device(device)
          logging.info("Device was created:\n%r\n", device)

          relation = EntityRelation(_from=asset.id, to=device.id, type="Contains")
          rest_client.save_relation(relation)
          logging.info("Relation was created:\n%r\n", relation)
      except ApiException as e:
          logging.exception(e)

if __name__ == '__main__':
  main()

Manage devices

Create a device, save and read attributes, then delete:

import logging
from tb_rest_client.rest_client_ce import *
from tb_rest_client.rest import ApiException

logging.basicConfig(level=logging.DEBUG,
                  format='%(asctime)s - %(levelname)s - %(module)s - %(lineno)d - %(message)s',
                  datefmt='%Y-%m-%d %H:%M:%S')

url = "https://eu.thingsboard.cloud"
username = "[email protected]"
password = "tenant"

def main():
  with RestClientCE(base_url=url) as rest_client:
      try:
          rest_client.login(username=username, password=password)
          default_device_profile_id = rest_client.get_default_device_profile_info().id
          device = Device(name="Thermometer 1",
                          device_profile_id=default_device_profile_id)
          device = rest_client.save_device(device)
          logging.info("Device was created:\n%r\n", device)

          found_device = rest_client.get_device_by_id(DeviceId(device.id, 'DEVICE'))

          rest_client.save_device_attributes(DeviceId(device.id, 'DEVICE'), 'SERVER_SCOPE',
                                             {'targetTemperature': 22.4})

          res = rest_client.get_attributes_by_scope(EntityId(device.id, 'DEVICE'), 'SERVER_SCOPE',
                                                    'targetTemperature')
          logging.info("Found device attributes: \n%r", res)

          rest_client.delete_device(DeviceId(device.id, 'DEVICE'))
      except ApiException as e:
          logging.exception(e)

if __name__ == '__main__':
  main()

Fetch tenant devices

import logging
from tb_rest_client.rest_client_ce import *
from tb_rest_client.rest import ApiException

logging.basicConfig(level=logging.DEBUG,
                  format='%(asctime)s - %(levelname)s - %(module)s - %(lineno)d - %(message)s',
                  datefmt='%Y-%m-%d %H:%M:%S')

url = "https://eu.thingsboard.cloud"
username = "[email protected]"
password = "tenant"

def main():
  with RestClientCE(base_url=url) as rest_client:
      try:
          rest_client.login(username=username, password=password)
          res = rest_client.get_tenant_devices(page_size=10, page=0)
          logging.info("Device info:\n%r", res)
      except ApiException as e:
          logging.exception(e)

if __name__ == '__main__':
  main()

Count entities with Entity Data Query API

import logging
from tb_rest_client.rest_client_ce import *
from tb_rest_client.rest import ApiException

logging.basicConfig(level=logging.DEBUG,
                  format='%(asctime)s - %(levelname)s - %(module)s - %(lineno)d - %(message)s',
                  datefmt='%Y-%m-%d %H:%M:%S')

url = "https://eu.thingsboard.cloud"
username = "[email protected]"
password = "tenant"

with RestClientCE(base_url=url) as rest_client:
  try:
      rest_client.login(username=username, password=password)
      entity_count_query = EntityCountQuery(
          entity_filter=EntityFilter(type='entityType', entity_type='DEVICE'),
          key_filters=[KeyFilter(
              key={"type": "ATTRIBUTE", "key": "active"},
              value_type='BOOLEAN',
              predicate={"operation": "EQUAL",
                         "value": {"defaultValue": True, "dynamicValue": None},
                         "type": "BOOLEAN"})])
      count = rest_client.count_entities_by_query(entity_count_query)
      logging.info("Total active devices: \n%r", count)
  except ApiException as e:
      logging.exception(e)

Version control scripts

The Python REST Client includes helper scripts for saving and loading entities to/from version control (ThingsBoard 3.4+).

Configure VCS access

python3 configure_vcs_access.py -H YOUR_THINGSBOARD_HOST -p YOUR_THINGSBOARD_PORT \
    -U YOUR_USER_EMAIL -P YOUR_USER_PASSWORD \
    -r YOUR_REPOSITORY_URL -b DEFAULT_BRANCH \
    -gu YOUR_VCS_USERNAME -gp YOUR_VCS_ACCESS_TOKEN

For private key authentication, use -pk and -pkp instead of -gp.

Save entities to VCS

wget https://raw.githubusercontent.com/thingsboard/thingsboard-python-rest-client/master/examples/load_all_entities_to_vcs_ce.py

python3 load_all_entities_to_vcs_ce.py -H YOUR_THINGSBOARD_HOST -p YOUR_THINGSBOARD_PORT \
    -U YOUR_USER_EMAIL -P YOUR_USER_PASSWORD

Load entities from VCS

wget https://raw.githubusercontent.com/thingsboard/thingsboard-python-rest-client/master/examples/load_all_entities_from_vcs_ce.py

python3 load_all_entities_from_vcs_ce.py -H YOUR_THINGSBOARD_HOST -p YOUR_THINGSBOARD_PORT \
    -U YOUR_USER_EMAIL -P YOUR_USER_PASSWORD -N YOUR_VERSION_NAME

More examples

See the thingsboard-python-rest-client repository for complete examples.

Was this helpful?