Docs / Contact / sdks/python/examples

Python SDK: Examples

Practical examples for common use cases with the Tendrl Python SDK.

Periodic Sensor Collection#

Use the @tether decorator to automatically publish the return value of a function:

python


from tendrl import Client

client = Client(api_key="your_key", offline_storage=True)
client.start()

@client.tether(tags=["sensors", "environment"])
def read_sensors():
    # Replace with actual sensor reading logic
    return {
        "temperature": 23.5,
        "humidity": 60,
        "pressure": 1013.25
    }

try:
    while True:
        read_sensors()  # Auto-publishes the return value
        time.sleep(10)
except KeyboardInterrupt:
    client.stop()

Tether with Offline Backup#

python


@client.tether(tags=["critical"], write_offline=True, db_ttl=3600)
def critical_reading():
    return {"voltage": 3.3, "current": 0.5}

When write_offline=True, the message is also stored locally. If the network send fails, the offline copy ensures delivery when connectivity returns. db_ttl sets the storage expiration in seconds.

Raspberry Pi GPIO Sensors#

python


from tendrl import Client

try:
    from gpiozero import CPUTemperature, DiskUsage
    cpu = CPUTemperature()
    disk = DiskUsage()
    simulated = False
except ImportError:
    simulated = True

client = Client(
    api_key="your_key",
    offline_storage=True,
    debug=True
)
client.start()

try:
    while True:
        if simulated:
            data = {"cpu_temp": 45.0, "disk_usage": 30.0, "simulated": True}
        else:
            data = {
                "cpu_temp": cpu.temperature,
                "disk_usage": disk.usage * 100
            }

        client.publish(data, tags=["rpi", "system"])
        time.sleep(30)
except KeyboardInterrupt:
    client.stop()

Handling Inbound Messages#

The Python SDK has no routing decorators. Pass one callback to the constructor and dispatch on the message yourself — a dict keyed by tag keeps it tidy:

python


from tendrl import Client


def run_diagnostics(message):
    report = {"self_test": "pass", "requested": message.get("data", {})}
    client.publish(report, tags=["diagnostic-result"])


def handle_ai_reply(message):
    print("AI:", message.get("data", {}).get("response"))


def handle_alert(message):
    print("Alert:", message.get("data"))


ROUTES = {
    "diagnostic": run_diagnostics,
    "ai-response": handle_ai_reply,
    "alert": handle_alert,
    "anomaly": handle_alert,
}


def on_message(message):
    for tag in message.get("tags") or []:
        handler = ROUTES.get(tag)
        if handler:
            handler(message)
            return
    print("No route:", message.get("msg_type"), message.get("tags"))


client = Client(api_key="your_key", callback=on_message)
client.start()

Polling only runs while a callback is set, so passing one is what turns receiving on at all. Tune it with check_msg_rate (seconds between polls) and check_msg_limit (messages per poll).

State Table Management#

The Python SDK doesn't currently wrap the state table, so use the REST endpoints directly. Authenticate with the entity's API key in Authorization: Bearer ....

python



API_KEY = os.environ["TENDRL_KEY"]
APP_URL = os.getenv("TENDRL_APP_URL", "https://app.tendrl.com").rstrip("/")
BASE = f"{APP_URL}/api/entities"
headers = {"Authorization": f"Bearer {API_KEY}"}

# Read current state
state = httpx.get(f"{BASE}/status-table", headers=headers).json()
print(f"Current state: {state}")

# Merge new fields (PATCH; existing keys preserved)
httpx.patch(
    f"{BASE}/status-table",
    headers=headers,
    json={"firmware_version": "2.1.0", "last_boot": "2025-01-15T10:30:00Z"},
)

# Replace entire state (PUT; all previous keys removed)
httpx.put(
    f"{BASE}/status-table",
    headers=headers,
    json={"firmware_version": "2.1.0", "status": "active", "config": {"interval": 30}},
)

Data Types#

publish() takes a dict or a string — nothing else. A list, a number or a tuple raises ValueError: Invalid type: <class 'list'>. Wrap those in a dict:

python


# Dictionary (most common)
client.publish({"key": "value"})

# String (auto-wrapped as {"data": "..."})
client.publish("simple message")

# A list must be wrapped in a dict; client.publish([1, 2, 3]) raises ValueError
client.publish({"readings": [1, 2, 3]})

# Nested structures
client.publish({
    "sensors": [
        {"id": "temp-1", "value": 23.5},
        {"id": "temp-2", "value": 24.1}
    ],
    "timestamp": "2025-01-15T10:00:00Z"
})

Whatever you pass must be JSON-serializable, since the dict is serialized as-is.

← Back to Contact docs