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()

Inbound Message Routing

Route incoming messages by tag or msg_type with @client.on():

python


from tendrl import Client

client = Client(api_key="your_key")

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

@client.on(tag="ai-response")
def handle_ai_reply(message):
    print("AI:", message.get("data", {}).get("response"))

@client.on(tags=["alert", "anomaly"])
def handle_alert(message):
    print("Alert:", message.get("data"))

@client.on_default
def unhandled(message):
    print("No route:", message.get("msg_type"), message.get("tags"))

client.start()

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"]
BASE = "https://app.tendrl.com/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

The SDK accepts any JSON-serializable data:

python


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

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

# List
client.publish([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"
})

← Back to Contact docs