Python REST Client (Legacy)

The ThingsBoard Python REST Client wraps the Edge REST API and lets Python scripts manage devices, assets, dashboards, customers, and other entities on your Edge node. It supports context managers for automatic token refresh and connects directly to the Edge node’s local REST endpoint.

Installation

pip3 install tb-rest-client==4.3.1.6

Authentication

Point the client at your Edge node’s local address — default http://localhost:8080.

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 = "http://localhost:8080"
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 = "http://localhost:8080"
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 = "http://localhost:8080"
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 = "http://localhost:8080"
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 = "http://localhost:8080"
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)

More examples

For complete examples, see the thingsboard-python-rest-client repository.

Was this helpful?