Advanced Actions Usage

This guide covers advanced patterns and deep dives into the Zelos Actions API, including complex field types, validation patterns, dynamic behavior, and integration techniques.

Field Types Deep Dive

Text Fields with Validation

Text fields support rich validation including length constraints and pattern matching:

from zelos_sdk import action

@action("Create User", "Create a new user account")
@action.input("username", type="text", required=True,
              min_length=3, max_length=20,
              pattern=r"^[a-zA-Z0-9_]+$",
              description="Username (letters, numbers, underscore only)")
@action.input("email", type="email", required=True,
              description="Valid email address")
@action.input("bio", type="text", max_length=500,
              description="Optional user biography")
def create_user(username: str, email: str, bio: str = ""):
    return {
        "user_id": f"user_{username}",
        "profile": {
            "username": username,
            "email": email,
            "bio": bio,
            "created": True
        }
    }

Number Fields with Constraints

Number fields support precise minimum/maximum validation and different UI widgets:

@action("Configure Motor", "Set motor control parameters")
@action.input("speed_rpm", type="number", minimum=0, maximum=3000,
              required=True, title="Speed (RPM)")
@action.input("torque_percent", type="number", minimum=0, maximum=100,
              default=50, title="Torque (%)")
@action.input("acceleration", type="number", minimum=0.1, maximum=10.0,
              default=1.0, title="Acceleration (m/s²)")
@action.input("enable_brake", type="boolean", default=False,
              title="Enable Regenerative Braking")
def configure_motor(speed_rpm: float, torque_percent: float = 50,
                   acceleration: float = 1.0, enable_brake: bool = False):
    config = {
        "motor_config": {
            "speed_setpoint": speed_rpm,
            "torque_limit": torque_percent,
            "acceleration_rate": acceleration,
            "regen_braking": enable_brake
        },
        "safety_checks": {
            "speed_valid": 0 <= speed_rpm <= 3000,
            "torque_valid": 0 <= torque_percent <= 100
        }
    }
    return config

Advanced Select Fields

Select fields can have static or dynamic choices, with dependency chains:

VEHICLE_TYPES = ["electric", "hybrid", "ice"]

def get_battery_types(vehicle_type: str):
    battery_types = {
        "electric": ["lithium_ion", "solid_state", "lithium_polymer"],
        "hybrid": ["nickel_metal_hydride", "lithium_ion"],
        "ice": []
    }
    return battery_types.get(vehicle_type, [])

def get_charging_options(vehicle_type: str, battery_type: str):
    if vehicle_type == "ice":
        return []
    charging_options = {
        "lithium_ion": ["ac_level1", "ac_level2", "dc_fast"],
        "solid_state": ["ac_level2", "dc_fast", "wireless"],
        "lithium_polymer": ["ac_level1", "ac_level2"],
        "nickel_metal_hydride": ["ac_level1"]
    }
    return charging_options.get(battery_type, [])

@action("Configure Vehicle", "Set up vehicle configuration")
@action.input("vehicle_type", type="select", choices=VEHICLE_TYPES, required=True)
@action.input("battery_type", type="select", choices=get_battery_types,
              depends_on="vehicle_type", required=True)
@action.input("charging", type="select", choices=get_charging_options,
              depends_on=["vehicle_type", "battery_type"])
def configure_vehicle(vehicle_type: str, battery_type: str = None, charging: str = None):
    config = {
        "vehicle_type": vehicle_type,
        "powertrain": {
            "battery_type": battery_type,
            "charging_capability": charging
        },
        "compatibility_check": charging is not None if battery_type else True
    }
    return config

Array Fields and Complex Structures

Handle arrays of items with validation:

@action("Configure Sensor Array", "Set up multiple sensors")
@action.input(
    "sensors",
    type="array",
    item={
        "type": "object",
        "properties": {
            "id": {"type": "string"},
            "type": {"type": "string", "enum": ["temperature", "pressure", "flow"]},
            "sample_rate": {"type": "number", "minimum": 1, "maximum": 1000}
        },
        "required": ["id", "type"]
    },
    min_items=1,
    max_items=10,
    required=True
)
@action.input("enable_sync", type="boolean", default=False, description="Synchronize all sensor readings")
def configure_sensor_array(sensors: list, enable_sync: bool = False):
    total_rate = sum(sensor.get("sample_rate", 100) for sensor in sensors)
    return {
        "sensor_count": len(sensors),
        "sensors": sensors,
        "synchronized": enable_sync,
        "total_sample_rate": total_rate,
        "warnings": ["High sample rate may impact performance"] if total_rate > 5000 else []
    }

File Upload Fields

Handle single and multiple file uploads:

@action("Process Dataset", "Upload and process data files")
@action.input("config_file", type="file", required=True,
              accept=".json,.yaml,.yml",
              description="Configuration file (JSON or YAML)")
@action.input("data_files", type="files",
              accept=".csv,.txt,.json",
              description="Data files to process")
@action.input("output_format", type="select",
              choices=["json", "csv", "parquet"], default="json")
def process_dataset(config_file: str, data_files: list = None, output_format: str = "json"):
    import base64
    import json

    # Parse config file (data URL format)
    if config_file.startswith("data:"):
        # Extract and decode base64 content
        header, data = config_file.split(",", 1)
        config_content = base64.b64decode(data).decode()
        config = json.loads(config_content)
    else:
        config = {}

    file_count = len(data_files) if data_files else 0

    return {
        "config": config,
        "files_received": file_count,
        "output_format": output_format,
        "processing_status": "ready",
        "estimated_time": file_count * 2  # 2 seconds per file
    }

Complex Validation Patterns

Cross-Field Validation

Implement validation that spans multiple fields:

from zelos_sdk.actions.types import ValidationError

@action("Schedule Maintenance", "Schedule vehicle maintenance")
@action.input("start_date", type="date", required=True)
@action.input("end_date", type="date", required=True)
@action.input("service_type", type="select",
              choices=["oil_change", "brake_service", "battery_replacement"], required=True)
@action.input("priority", type="select", choices=["low", "medium", "high"], default="medium")
def schedule_maintenance(start_date: str, end_date: str, service_type: str, priority: str = "medium"):
    from datetime import datetime

    # Parse dates
    start = datetime.strptime(start_date, "%Y-%m-%d")
    end = datetime.strptime(end_date, "%Y-%m-%d")

    # Cross-field validation
    if end <= start:
        raise ValidationError("end_date", "End date must be after start date", "date_range")

    # Business logic validation
    duration = (end - start).days
    min_duration = {"oil_change": 1, "brake_service": 2, "battery_replacement": 3}

    if duration < min_duration[service_type]:
        raise ValidationError(
            "end_date",
            f"{service_type} requires at least {min_duration[service_type]} days",
            "insufficient_duration"
        )

    return {
        "maintenance_id": f"maint_{start_date}_{service_type}",
        "schedule": {
            "start": start_date,
            "end": end_date,
            "duration_days": duration,
            "service": service_type,
            "priority": priority
        }
    }

Custom Field Types

Create reusable custom field types:

from zelos_sdk.actions.types import FieldType
from zelos_sdk.actions.fields import BaseField, register_field_type, ValidationError
import re

@register_field_type("vin")
class VINField(BaseField):
    """Vehicle Identification Number field with validation"""

    def __init__(self, name: str, **kwargs):
        super().__init__(name, **kwargs)
        self.field_type = FieldType.STRING

    def validate(self, value, form_data=None):
        value = super().validate(value, form_data)
        if value is not None:
            # VIN validation rules
            if len(value) != 17:
                raise ValidationError(self.name, "VIN must be exactly 17 characters", "length")

            if not re.match(r"^[A-HJ-NPR-Z0-9]{17}$", value):
                raise ValidationError(self.name, "Invalid VIN format", "format")

            # Check digits and other VIN-specific validations...

        return value

# Use the custom field
@action("Register Vehicle", "Register a new vehicle")
@action.input("vin", type="vin", required=True, description="17-character VIN")
@action.input("make", type="text", required=True)
@action.input("model", type="text", required=True)
@action.input("year", type="number", minimum=1900, maximum=2030, required=True)
def register_vehicle(vin: str, make: str, model: str, year: int):
    return {
        "vehicle_id": f"v_{vin[-8:]}",
        "registration": {
            "vin": vin,
            "make": make,
            "model": model,
            "year": year,
            "status": "registered"
        }
    }

Class-Based Action Patterns

Stateful Action Classes

Maintain state across action executions:

# Simple stateful action class example
from zelos_sdk import action, actions_registry
import time

# Global variable to hold the task manager instance
task_manager = None

def get_task_choices():
    """Get list of available task IDs for dropdown"""
    global task_manager
    if task_manager is None:
        return []
    return list(task_manager._tasks.keys())

class TaskManager:
    """Simple task manager that maintains state across actions"""

    # Class-level storage for the singleton instance
    _instance = None
    _tasks = {}
    _task_counter = 0

    def __init__(self):
        # Ensure we only create one instance
        if TaskManager._instance is not None:
            raise RuntimeError("TaskManager is a singleton")
        TaskManager._instance = self

    @classmethod
    def get_instance(cls):
        """Get the singleton instance"""
        if cls._instance is None:
            cls._instance = cls()
        return cls._instance

    @action("Create Task", "Create a new task")
    @action.input("title", type="text", required=True, min_length=1, max_length=100)
    @action.input("priority", type="select", choices=["low", "medium", "high"], default="medium")
    @action.input("description", type="text", max_length=500, default="")
    def create_task(self, title: str, priority: str = "medium", description: str = ""):
        TaskManager._task_counter += 1
        task_id = f"task_{TaskManager._task_counter}"

        TaskManager._tasks[task_id] = {
            "title": title,
            "priority": priority,
            "description": description,
            "status": "pending",
            "created_at": time.time()
        }

        return {
            "task_id": task_id,
            "task": TaskManager._tasks[task_id],
            "message": "Task created successfully"
        }

    @action("List Tasks", "Get all tasks")
    def list_tasks(self):
        return {
            "tasks": TaskManager._tasks,
            "count": len(TaskManager._tasks)
        }

    @action("Update Task Status", "Update task status")
    @action.input("task_id", type="select", choices=get_task_choices, required=True)
    @action.input("status", type="select", choices=["pending", "in_progress", "completed", "cancelled"], required=True)
    def update_task_status(self, task_id: str, status: str):
        if task_id not in TaskManager._tasks:
            return {"error": "Task not found"}

        TaskManager._tasks[task_id]["status"] = status
        return {
            "task_id": task_id,
            "task": TaskManager._tasks[task_id],
            "message": f"Task status updated to {status}"
        }

    @action("Delete Task", "Delete a task")
    @action.input("task_id", type="select", choices=get_task_choices, required=True)
    def delete_task(self, task_id: str):
        if task_id not in TaskManager._tasks:
            return {"error": "Task not found"}

        deleted_task = TaskManager._tasks.pop(task_id)
        return {
            "task_id": task_id,
            "deleted_task": deleted_task,
            "message": "Task deleted successfully"
        }

# Create and register the task manager
task_manager = TaskManager()
actions_registry.register(task_manager, name="tasks")

Inheritance and Action Composition

Use inheritance to create action hierarchies:

class BaseDeviceController:
    """Base class for device controllers"""

    def __init__(self, device_type: str):
        self.device_type = device_type
        self.devices = {}

    @action("List Devices", "List all connected devices")
    def list_devices(self):
        return {
            "device_type": self.device_type,
            "count": len(self.devices),
            "devices": list(self.devices.keys())
        }

class MotorController(BaseDeviceController):
    def __init__(self):
        super().__init__("motor")

    @action("Start Motor", "Start a motor by ID")
    @action.input("motor_id", type="text", required=True)
    @action.input("speed_rpm", type="number", minimum=0, maximum=3000, default=1000)
    def start_motor(self, motor_id: str, speed_rpm: float = 1000):
        self.devices[motor_id] = {"status": "running", "speed": speed_rpm}
        return {"motor_id": motor_id, "status": "started", "speed": speed_rpm}

class SensorController(BaseDeviceController):
    def __init__(self):
        super().__init__("sensor")

    @action("Read Sensor", "Read current sensor value")
    @action.input("sensor_id", type="text", required=True)
    def read_sensor(self, sensor_id: str):
        import random
        value = random.uniform(20, 25)  # Mock temperature reading
        self.devices[sensor_id] = {"last_reading": value, "timestamp": time.time()}
        return {"sensor_id": sensor_id, "value": value, "unit": "celsius"}

# Create instances
motor_ctrl = MotorController()
sensor_ctrl = SensorController()

Error Handling and Response Patterns

Structured Error Responses

Return consistent error formats:

from zelos_sdk.actions.types import ActionResult

@action("Transfer Funds", "Transfer money between accounts")
@action.input("from_account", type="text", required=True)
@action.input("to_account", type="text", required=True)
@action.input("amount", type="number", minimum=0.01, required=True)
@action.input("currency", type="select", choices=["USD", "EUR", "GBP"], default="USD")
def transfer_funds(from_account: str, to_account: str, amount: float, currency: str = "USD"):
    # Simulate account validation
    valid_accounts = ["acc_001", "acc_002", "acc_003"]

    if from_account not in valid_accounts:
        return ActionResult.error({
            "error_code": "INVALID_ACCOUNT",
            "message": f"Account {from_account} not found",
            "field": "from_account",
            "valid_accounts": valid_accounts
        })

    if to_account not in valid_accounts:
        return ActionResult.error({
            "error_code": "INVALID_ACCOUNT",
            "message": f"Account {to_account} not found",
            "field": "to_account"
        })

    if from_account == to_account:
        return ActionResult.error({
            "error_code": "SAME_ACCOUNT",
            "message": "Cannot transfer to the same account",
            "suggestion": "Please select different accounts"
        })

    # Simulate balance check
    balances = {"acc_001": 1000, "acc_002": 500, "acc_003": 2000}
    if balances.get(from_account, 0) < amount:
        return ActionResult.error({
            "error_code": "INSUFFICIENT_FUNDS",
            "message": "Insufficient balance",
            "available_balance": balances.get(from_account, 0),
            "requested_amount": amount
        })

    # Success case
    transaction_id = f"txn_{int(time.time())}"
    return ActionResult.success({
        "transaction_id": transaction_id,
        "from_account": from_account,
        "to_account": to_account,
        "amount": amount,
        "currency": currency,
        "status": "completed",
        "new_balance": balances[from_account] - amount
    })

Progressive Enhancement

Actions that provide different levels of detail:

@action("System Diagnostics", "Run comprehensive system diagnostics")
@action.input("detail_level", type="select",
              choices=["basic", "detailed", "comprehensive"], default="basic")
@action.input("include_history", type="boolean", default=False)
@action.input("export_format", type="select",
              choices=["json", "xml", "csv"], default="json")
def system_diagnostics(detail_level: str = "basic", include_history: bool = False, export_format: str = "json"):
    import random

    # Basic diagnostics always included
    basic_info = {
        "system_status": "healthy",
        "uptime_hours": random.randint(1, 168),
        "cpu_usage": round(random.uniform(10, 80), 1),
        "memory_usage": round(random.uniform(30, 90), 1)
    }

    result = {"basic": basic_info}

    if detail_level in ["detailed", "comprehensive"]:
        result["detailed"] = {
            "network_status": "connected",
            "disk_usage": round(random.uniform(20, 95), 1),
            "active_processes": random.randint(50, 200),
            "error_count": random.randint(0, 5)
        }

    if detail_level == "comprehensive":
        result["comprehensive"] = {
            "hardware_info": {
                "cpu_temp": round(random.uniform(40, 70), 1),
                "fan_speed": random.randint(1000, 3000),
                "voltage_levels": {"12v": 12.1, "5v": 5.0, "3.3v": 3.28}
            },
            "performance_metrics": {
                "avg_response_time": round(random.uniform(10, 100), 2),
                "throughput_ops_sec": random.randint(100, 1000)
            }
        }

    if include_history:
        result["history"] = {
            "last_24_hours": [
                {"hour": i, "cpu": round(random.uniform(10, 80), 1)}
                for i in range(24)
            ]
        }

    result["export_format"] = export_format
    result["generated_at"] = time.time()

    return result

Entry Points

Entry points allow your actions to be automatically discovered and registered when the SDK is imported.

Setup

Add an entry point to your pyproject.toml:

[project.entry-points."zelos_sdk.actions"]
my_actions = "my_package.actions"

Example

Create your actions module:

# my_package/actions.py
from zelos_sdk import action

@action("Process Data", "Process incoming data")
def echo(data: str) -> dict:
    return {"message": data, data}

@action("Calculate Sum", "Add numbers together")
@action.input("numbers", type="array", required=True)
def calculate_sum(numbers: list) -> dict:
    return {"sum": sum(numbers)}

Your actions are now discoverable:

from zelos_sdk import actions_registry

# Actions are automatically registered
actions = actions_registry.list()
print(actions)  # ['my_actions/echo', 'my_actions/calculate_sum']

# Execute actions
result = actions_registry.execute("my_actions/echo", {"data": "hello"})
print(result)  # {'message': 'hello'}

That's it! Your actions are now discoverable by the Zelos SDK.