Python SDK

from lox_backup import LoxClient

# Initialize client
client = LoxClient(api_key="your-api-key")

# Upload a backup
backup = client.backups.upload(
    file_path="backup.tar.gz",
    name="my-backup",
    tags=["production", "database"],
    retention_days=30
)

print(f"Backup created: {backup.uuid}")
print(f"Status: {backup.status}")
print(f"Size: {backup.size_bytes} bytes")

from lox_backup import LoxClient

client = LoxClient(
    api_key="your-api-key",      # Required: Your LOX API key
    base_url="https://backlox.com/api", # Optional: API base URL
    timeout=300,                  # Optional: Request timeout in seconds
    max_retries=3,               # Optional: Max retry attempts
)

export LOX_API_KEY="your-api-key"
export LOX_API_URL="https://backlox.com/api"  # Optional

from lox_backup import LoxClient

# API key is read from LOX_API_KEY environment variable
client = LoxClient()

from lox_backup import LoxClient

client = LoxClient()

# Simple upload
backup = client.backups.upload("backup.tar.gz")

# Upload with options
backup = client.backups.upload(
    file_path="backup.tar.gz",
    name="production-db-2024-01-15",
    description="Daily production database backup",
    tags=["production", "database", "mysql"],
    retention_days=90,
    wait_for_completion=True,  # Wait for processing to complete
    timeout=3600               # Max wait time in seconds
)

print(f"UUID: {backup.uuid}")
print(f"Status: {backup.status}")
print(f"Size: {backup.size_bytes}")

# List all backups
backups = client.backups.list()

for backup in backups:
    print(f"{backup.name}: {backup.status}")

# Filter by status
completed = client.backups.list(status="completed")

# Filter by tags
production = client.backups.list(tags="production")

# Search by name
results = client.backups.list(search="database")

# Pagination
page1 = client.backups.list(limit=10, offset=0)
page2 = client.backups.list(limit=10, offset=10)

# Get backup by UUID
backup = client.backups.get("550e8400-e29b-41d4-a716-446655440000")

print(f"Name: {backup.name}")
print(f"Status: {backup.status}")
print(f"Size: {backup.size_bytes}")
print(f"Created: {backup.created_at}")
print(f"Tags: {backup.tags}")

# Request restore (generates download URL)
restore = client.backups.restore("550e8400-e29b-41d4-a716-446655440000")

print(f"Download URL: {restore.download_url}")
print(f"Expires: {restore.expires_at}")

# Download to file
client.backups.download(
    uuid="550e8400-e29b-41d4-a716-446655440000",
    output_path="restored-backup.tar.gz"
)

# Delete a backup
client.backups.delete("550e8400-e29b-41d4-a716-446655440000")

print("Backup deleted")

# Upload without waiting
backup = client.backups.upload("backup.tar.gz", wait_for_completion=False)
print(f"Upload started: {backup.uuid}")

# Wait for completion later
completed = client.backups.wait_for_completion(
    uuid=backup.uuid,
    timeout=3600,      # Max wait time
    poll_interval=5    # Check every 5 seconds
)

print(f"Backup completed: {completed.status}")

tenant = client.tenant.get_current()

print(f"Name: {tenant.name}")
print(f"Email: {tenant.email}")
print(f"Created: {tenant.created_at}")

quota = client.tenant.get_quota()

print(f"Used: {quota.used_bytes} bytes")
print(f"Quota: {quota.quota_bytes} bytes")
print(f"Usage: {quota.usage_percentage}%")
print(f"Unlimited: {quota.unlimited}")

# Check if storage is available
if client.tenant.has_available_storage(required_bytes=1024*1024*100):
    print("100MB available")
else:
    print("Not enough storage")

targets = client.storage.list()

for target in targets:
    print(f"{target.name}: {target.provider} - {target.status}")

health = client.storage.get_health_summary()

print(f"Healthy: {health.healthy_count}")
print(f"Degraded: {health.degraded_count}")
print(f"Offline: {health.offline_count}")

# List all notification channels
channels = client.notification_channels.list()

for channel in channels:
    print(f"{channel.name}: {channel.channel_type} - {'enabled' if channel.is_enabled else 'disabled'}")

# Create a webhook channel
channel = client.notification_channels.create(
    name="Backup Alerts",
    channel_type="webhook",
    config={"url": "https://api.example.com/webhooks/lox"},
    events=["backup.completed", "backup.failed", "backup.quarantine"]
)

# Create a Slack channel
slack = client.notification_channels.create(
    name="Slack Notifications",
    channel_type="slack",
    config={"webhook_url": "https://hooks.slack.com/services/xxx"}
)

# Test a channel
result = client.notification_channels.test(channel.uuid)
print(f"Test {'passed' if result.success else 'failed'}: {result.message}")

# Enable/disable channel
client.notification_channels.enable(channel.uuid)
client.notification_channels.disable(channel.uuid)

# Get channel summary
summary = client.notification_channels.get_summary()
print(f"Total: {summary.total_channels}, Healthy: {summary.healthy_channels}")

# Register a new agent
registration = client.agents.register(
    hostname="server-01.example.com",
    os_type="linux",
    display_name="Production Server 01",
    description="Main production database server"
)

print(f"Agent UUID: {registration.uuid}")
print(f"Token: {registration.registration_token}")  # Save this securely!

# List all agents
agents = client.agents.list()
for agent in agents:
    print(f"{agent.hostname}: {agent.status}")

# Get online agents only
online = client.agents.list_online()

# Get agent details
agent = client.agents.get("agent-uuid")
print(f"Status: {agent.status}, Backups: {agent.total_backups}")

# Get installation command
cmd = client.agents.get_install_command("agent-uuid")
print(f"Install: {cmd.recommended_command}")

# Get agent summary
summary = client.agents.get_summary()
print(f"Online: {summary.online_agents}/{summary.total_agents}")

# Enable/disable agent
client.agents.enable("agent-uuid")
client.agents.disable("agent-uuid")

# Create a backup job
job = client.jobs.create(
    agent_uuid="agent-uuid",
    name="Daily Database Backup",
    job_type="database",
    sources=[{"path": "/var/lib/mysql", "recursive": True}],
    schedule_enabled=True,
    schedule_cron="0 2 * * *",  # Daily at 2 AM
    retention_policy={
        "keep_daily": 7,
        "keep_weekly": 4,
        "keep_monthly": 12
    }
)

# List all jobs
jobs = client.jobs.list()
for job in jobs:
    print(f"{job.name}: {job.last_status}")

# Run a job immediately
result = client.jobs.run("job-uuid", backup_type="full")
print(f"Backup started: {result.backup_uuid}")

# List backups for a job
backups = client.jobs.list_backups("job-uuid")
for backup in backups:
    print(f"{backup.backup_type}: {backup.status} - {backup.size_bytes} bytes")

# Enable/disable job
client.jobs.enable("job-uuid")
client.jobs.disable("job-uuid")

# List machine backups
backups = client.machine_backups.list(status="completed", limit=10)

# Get backup details
backup = client.machine_backups.get("backup-uuid")
print(f"Size: {backup.size_bytes}, Files: {backup.files_count}")

# Restore a backup
restore = client.machine_backups.restore(
    backup_uuid="backup-uuid",
    restore_type="full",
    target_type="original",
    overwrite_existing=False
)
print(f"Restore job: {restore.uuid}")

# Delete a backup
client.machine_backups.delete("backup-uuid")

from lox_backup import LoxClient
from lox_backup.exceptions import (
    LoxError,
    AuthenticationError,
    NotFoundError,
    QuotaExceededError,
    ValidationError,
    RateLimitError,
)

client = LoxClient()

try:
    backup = client.backups.upload("backup.tar.gz")
except AuthenticationError:
    print("Invalid API key")
except QuotaExceededError:
    print("Storage quota exceeded")
except NotFoundError:
    print("Resource not found")
except ValidationError as e:
    print(f"Validation error: {e.message}")
except RateLimitError:
    print("Rate limited, try again later")
except LoxError as e:
    print(f"API error: {e.message}")

import asyncio
from lox_backup import AsyncLoxClient

async def main():
    client = AsyncLoxClient(api_key="your-api-key")

    # Upload backup
    backup = await client.backups.upload(
        file_path="backup.tar.gz",
        name="async-backup"
    )

    print(f"Uploaded: {backup.uuid}")

    # List backups
    backups = await client.backups.list()
    for b in backups:
        print(f"  - {b.name}")

    await client.close()

asyncio.run(main())

#!/usr/bin/env python3
"""
LOX Backup - Database Backup Script
"""
import subprocess
import tempfile
import os
from datetime import datetime
from lox_backup import LoxClient
from lox_backup.exceptions import QuotaExceededError

def backup_database():
    client = LoxClient()

    # Check storage quota
    quota = client.tenant.get_quota()
    print(f"Storage used: {quota.usage_percentage}%")

    # Create database dump
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    with tempfile.TemporaryDirectory() as tmpdir:
        dump_file = os.path.join(tmpdir, f"db_{timestamp}.sql.gz")

        # Dump MySQL database
        cmd = f"mysqldump -u root mydb | gzip > {dump_file}"
        subprocess.run(cmd, shell=True, check=True)

        # Upload to LOX
        try:
            backup = client.backups.upload(
                file_path=dump_file,
                name=f"mysql-mydb-{timestamp}",
                tags=["mysql", "production", "automated"],
                retention_days=30,
                wait_for_completion=True
            )

            print(f"Backup complete!")
            print(f"  UUID: {backup.uuid}")
            print(f"  Size: {backup.size_bytes:,} bytes")
            print(f"  Expires: {backup.expires_at}")

        except QuotaExceededError:
            print("ERROR: Storage quota exceeded!")
            print("Please delete old backups or upgrade your plan.")
            return False

    return True

if __name__ == "__main__":
    backup_database()