unrealon 0.1.3

Unrealon SDK - Service management for Django backend (registration, heartbeat, logging, commands)

Unrealon SDK

Service management SDK for Unrealon platform. Provides registration, heartbeat, logging, and command handling.

Installation

pip install unrealon

Quick Start

from unrealon import ServiceClient, ServiceStatus

with ServiceClient(
    api_key="dk_xxx",
    service_name="my-service",
) as client:

    # Handle commands from UI
    client.on_command("pause", lambda cmd: client.update_status(status=ServiceStatus.PAUSED))
    client.on_command("resume", lambda cmd: client.update_status(status=ServiceStatus.RUNNING))
    client.on_command("stop", lambda cmd: client.request_shutdown())

    # Main loop
    while not client.shutdown_requested:
        process_item()
        client.increment_processed()

Configuration

Environment Variables

export UNREALON_API_KEY=dk_xxx
export UNREALON_SERVICE_NAME=my-service

Direct Configuration

client = ServiceClient(
    api_key="dk_xxx",
    service_name="my-service",
    source_code="parser",       # Optional identifier
    heartbeat_interval=30,      # Seconds between heartbeats
)

Settings Reference

Env Variable	Default	Description
UNREALON_API_KEY	required	API key
UNREALON_SERVICE_NAME	required	Service identifier
UNREALON_API_URL	https://api.unrealon.com	API endpoint
UNREALON_HEARTBEAT_INTERVAL	30	Heartbeat interval (seconds)
UNREALON_LOG_BATCH_SIZE	50	Max logs per batch
UNREALON_LOG_FLUSH_INTERVAL	5.0	Log flush interval (seconds)
UNREALON_COMMAND_POLL_INTERVAL	10	Command poll interval (seconds)

Features

Registration & Lifecycle

# Context manager (recommended)
with ServiceClient(api_key="...", service_name="...") as client:
    # Auto-registered on enter, auto-deregistered on exit
    pass

# Manual control
client = ServiceClient(api_key="...", service_name="...")
service_id = client.start()
# ... work ...
client.stop()

Status & Heartbeat

Heartbeats are sent automatically in background:

with ServiceClient(...) as client:
    # Check current status
    if client.status == ServiceStatus.PAUSED:
        time.sleep(1)
        continue

    # Update metrics (included in heartbeat)
    client.increment_processed(100)
    client.increment_errors(1)

    # Or update directly
    client.update_status(
        status=ServiceStatus.RUNNING,
        items_processed=1000,
        errors_count=5,
    )

Commands

Handle commands from server/UI:

from unrealon import ServiceClient, ServiceStatus

with ServiceClient(...) as client:

    def handle_pause(cmd):
        client.update_status(status=ServiceStatus.PAUSED)
        return {"paused": True}

    def handle_stop(cmd):
        client.request_shutdown()
        return {"stopping": True}

    client.on_command("pause", handle_pause)
    client.on_command("resume", lambda cmd: client.update_status(status=ServiceStatus.RUNNING))
    client.on_command("stop", handle_stop)

    while not client.shutdown_requested:
        if client.status == ServiceStatus.PAUSED:
            time.sleep(1)
            continue
        process_items()

Logging

Logs are batched and sent automatically:

with ServiceClient(...) as client:
    client.debug("Debug message")
    client.info("Info message", extra={"key": "value"})
    client.warning("Warning message")
    client.error("Error message")
    client.critical("Critical error")

Complete Example

#!/usr/bin/env python3
"""Example service using Unrealon SDK."""

import time
import random
from unrealon import ServiceClient, ServiceStatus


def run_service(api_key: str):
    with ServiceClient(
        api_key=api_key,
        service_name="example-service",
        source_code="example",
    ) as client:

        # Setup command handlers
        client.on_command("pause", lambda cmd: client.update_status(status=ServiceStatus.PAUSED))
        client.on_command("resume", lambda cmd: client.update_status(status=ServiceStatus.RUNNING))
        client.on_command("stop", lambda cmd: client.request_shutdown())

        print(f"Service registered: {client.service_id}")

        # Main loop
        while not client.shutdown_requested:
            # Skip work if paused
            if client.status == ServiceStatus.PAUSED:
                time.sleep(1)
                continue

            # Simulate work
            client.increment_processed(random.randint(1, 10))

            if random.random() < 0.05:
                client.increment_errors()

            time.sleep(1)

    print("Service stopped")


if __name__ == "__main__":
    import os
    run_service(api_key=os.environ["UNREALON_API_KEY"])

Service Status

from unrealon import ServiceStatus

# Available statuses:
ServiceStatus.INITIALIZING  # Starting up
ServiceStatus.RUNNING       # Normal operation
ServiceStatus.PAUSED        # Paused by command
ServiceStatus.STOPPING      # Shutting down
ServiceStatus.STOPPED       # Stopped
ServiceStatus.ERROR         # Error state
ServiceStatus.OFFLINE       # No heartbeat
ServiceStatus.STALE         # Heartbeat timeout

Error Handling

from unrealon import (
    UnrealonError,
    AuthenticationError,
    RegistrationError,
    NetworkError,
)

try:
    with ServiceClient(...) as client:
        pass
except AuthenticationError:
    print("Invalid API key")
except RegistrationError:
    print("Registration failed")
except NetworkError:
    print("Network error")
except UnrealonError:
    print("SDK error")

Async Support

from unrealon import AsyncServiceClient

async with AsyncServiceClient(
    api_key="...",
    service_name="...",
) as client:
    await client.info("Async message")

License

MIT