Skip to content

Key Expressions

Key expressions are the foundation of Skarv's message routing system. They allow you to organize and route messages using hierarchical patterns and wildcards.

Basic Syntax

Key expressions use a hierarchical structure with forward slashes (/) as separators:

level1/level2/level3

Examples

# Simple hierarchical keys
skarv.put("sensor/temperature", 22.5)
skarv.put("device/001/status", "online")
skarv.put("user/preferences/theme", "dark")

Pattern Matching

Skarv supports several wildcard patterns for flexible message routing:

Single Level Wildcard (*)

Matches exactly one level in the hierarchy:

# Subscribe to any device status
@skarv.subscribe("device/*/status")
def handle_device_status(sample):
    device_id = str(sample.key_expr).split('/')[1]
    print(f"Device {device_id}: {sample.value}")

# This will match:
skarv.put("device/001/status", "online")  # ✅
skarv.put("device/002/status", "offline") # ✅
skarv.put("device/status", "error")       # ❌ (missing level)
skarv.put("device/001/status/extra", "x") # ❌ (too many levels)

Multi-Level Wildcard (**)

Matches zero or more levels in the hierarchy:

# Subscribe to all sensor data
@skarv.subscribe("sensor/**")
def handle_all_sensors(sample):
    print(f"All sensors - {sample.key_expr}: {sample.value}")

# This will match:
skarv.put("sensor/temperature", 22.5)           # ✅
skarv.put("sensor/room1/temperature", 23.1)     # ✅
skarv.put("sensor/building/a/floor/2/temp", 24) # ✅
skarv.put("sensor", "general")                  # ✅

Mixed Patterns

You can combine exact matches with wildcards:

# Subscribe to temperature readings from any room
@skarv.subscribe("sensor/*/temperature")

# Subscribe to any sensor in room1
@skarv.subscribe("sensor/room1/*")

# Subscribe to all data from device 001
@skarv.subscribe("device/001/**")

Practical Examples

IoT Device Management

import skarv

# Device status monitoring
@skarv.subscribe("device/*/status")
def monitor_device_status(sample):
    device_id = str(sample.key_expr).split('/')[1]
    print(f"Device {device_id} status: {sample.value}")

# Temperature monitoring for all rooms
@skarv.subscribe("sensor/*/temperature")
def monitor_temperatures(sample):
    room = str(sample.key_expr).split('/')[1]
    print(f"Room {room} temperature: {sample.value}°C")

# All sensor data logging
@skarv.subscribe("sensor/**")
def log_all_sensors(sample):
    print(f"Sensor log: {sample.key_expr} = {sample.value}")

# Simulate device network
skarv.put("device/001/status", "online")
skarv.put("device/002/status", "offline")
skarv.put("sensor/living/temperature", 22.5)
skarv.put("sensor/bedroom/temperature", 21.8)
skarv.put("sensor/kitchen/humidity", 65.2)

Application Event System

import skarv

# User action events
@skarv.subscribe("user/*/action")
def handle_user_actions(sample):
    user_id = str(sample.key_expr).split('/')[1]
    print(f"User {user_id} performed action: {sample.value}")

# System events
@skarv.subscribe("system/**")
def handle_system_events(sample):
    print(f"System event: {sample.key_expr} - {sample.value}")

# Error logging
@skarv.subscribe("**/error")
def handle_errors(sample):
    print(f"Error in {sample.key_expr}: {sample.value}")

# Simulate application events
skarv.put("user/123/action", "login")
skarv.put("user/456/action", "logout")
skarv.put("system/startup", "Application started")
skarv.put("system/database/connection", "Connected")
skarv.put("user/123/error", "Invalid password")

Retrieving Data with Patterns

The skarv.get() function also supports pattern matching:

# Store some data
skarv.put("device/001/temperature", 23.5)
skarv.put("device/002/temperature", 24.1)
skarv.put("device/001/humidity", 65.2)
skarv.put("device/002/humidity", 68.1)

# Retrieve all temperatures
temperatures = skarv.get("device/*/temperature")
print(temperatures)
# Output: [Sample(key_expr='device/001/temperature', value=23.5),
#          Sample(key_expr='device/002/temperature', value=24.1)]

# Retrieve all data for device 001
device_001_data = skarv.put("device/001/*")
print(device_001_data)
# Output: [Sample(key_expr='device/001/temperature', value=23.5),
#          Sample(key_expr='device/001/humidity', value=65.2)]

# Retrieve all sensor data
all_sensors = skarv.get("**")
print(all_sensors)
# Output: All stored samples

Best Practices

1. Use Hierarchical Structure

Organize your keys in a logical hierarchy:

# Good
skarv.put("sensor/room1/temperature", 22.5)
skarv.put("sensor/room1/humidity", 65.2)
skarv.put("device/001/status", "online")

# Avoid flat structures
skarv.put("temperature_room1", 22.5)  # Less flexible

2. Be Specific with Subscriptions

Use specific patterns to avoid unintended matches:

# Specific subscription
@skarv.subscribe("sensor/*/temperature")  # Only temperature sensors

# Too broad
@skarv.subscribe("**")  # Everything - use with caution

3. Plan Your Key Structure

Design your key hierarchy before implementation:

application/
├── user/
│   ├── {user_id}/
│   │   ├── profile
│   │   ├── preferences
│   │   └── actions
├── system/
│   ├── status
│   ├── events
│   └── errors
└── data/
    ├── sensor/
    │   └── {sensor_id}/
    └── device/
        └── {device_id}/

4. Use Consistent Naming

Maintain consistent naming conventions:

# Use lowercase with underscores or hyphens
skarv.put("user_profile", {...})
skarv.put("system-status", "running")

# Avoid mixed case or special characters
skarv.put("UserProfile", {...})  # Inconsistent
skarv.put("system@status", "x")  # Special characters

Advanced Patterns

Conditional Subscriptions

You can use multiple subscriptions for complex routing:

# Handle different types of alerts
@skarv.subscribe("alert/critical/**")
def handle_critical_alerts(sample):
    print(f"🚨 CRITICAL: {sample.value}")

@skarv.subscribe("alert/warning/**")
def handle_warning_alerts(sample):
    print(f"⚠️  WARNING: {sample.value}")

@skarv.subscribe("alert/info/**")
def handle_info_alerts(sample):
    print(f"ℹ️  INFO: {sample.value}")

Data Aggregation

Use patterns to aggregate related data:

# Aggregate all temperature readings
@skarv.subscribe("sensor/*/temperature")
def aggregate_temperatures(sample):
    # Get all current temperatures
    all_temps = skarv.get("sensor/*/temperature")
    avg_temp = sum(s.value for s in all_temps) / len(all_temps)
    print(f"Average temperature: {avg_temp}°C")

Next Steps