Rastion Client
The rastion_client.py
module provides the core client library for integrating with the Rastion platform. It handles authentication, API communication, model management, and platform services integration.
Overview
The Rastion client enables:
- Authentication: Secure authentication with the Rastion platform
- Model Management: Upload, download, and manage optimization models
- API Access: Complete access to platform APIs and services
- Repository Integration: Seamless integration with platform repositories
- Cloud Services: Access to cloud execution and benchmarking
Core Classes
RastionClient
Main client class for platform interaction:
from qubots import RastionClient
# Initialize client
client = RastionClient(
api_base="https://hub.rastion.com/api/v1",
config_path="~/.rastion/config.json"
)
# Authenticate
client.authenticate(token="your_api_token")
# Check authentication status
if client.is_authenticated():
print(f"Authenticated as: {client.get_user_info()['username']}")
Authentication Methods
Multiple authentication options:
# Token-based authentication
client.authenticate(token="your_api_token")
# Interactive authentication
client.authenticate_interactive()
# Environment variable authentication
import os
os.environ['RASTION_API_TOKEN'] = 'your_token'
client.authenticate_from_env()
# Configuration file authentication
client.authenticate_from_config()
Model Management
Loading Models
Load optimization models from the platform:
import qubots.rastion as rastion
# Authenticate first
rastion.authenticate("your_api_token")
# Load models using convenience functions
problem = rastion.load_qubots_model("tsp_problem", username="community")
optimizer = rastion.load_qubots_model("genetic_tsp", username="algorithms")
# Use models locally
result = optimizer.optimize(problem)
print(f"Best solution: {result.best_value}")
Uploading Models
Upload your models to the platform:
from qubots import upload_qubots_model
# Upload a problem
upload_qubots_model(
model_path="./my_problem",
model_name="custom_tsp_problem",
model_type="problem",
description="Custom TSP problem with specific constraints",
tags=["tsp", "routing", "optimization"]
)
# Upload an optimizer
upload_qubots_model(
model_path="./my_optimizer",
model_name="hybrid_genetic_optimizer",
model_type="optimizer",
description="Hybrid genetic algorithm with local search",
tags=["genetic", "hybrid", "metaheuristic"]
)
Model Discovery
Search and discover models on the platform:
# Search for models
models = client.search_models(
query="tsp",
model_type="problem",
tags=["routing"],
username="community"
)
for model in models:
print(f"{model.name}: {model.description}")
print(f" Author: {model.username}")
print(f" Downloads: {model.download_count}")
print(f" Rating: {model.average_rating}/5")
Repository Integration
Repository Management
Manage your repositories on the platform:
# List your repositories
repos = client.list_repositories()
for repo in repos:
print(f"{repo.name}: {repo.description}")
# Create new repository
repo = client.create_repository(
name="my-optimization-suite",
description="Collection of optimization algorithms",
visibility="public"
)
# Clone repository locally
client.clone_repository("username/repository-name", "./local-path")
File Operations
Manage files within repositories:
# Upload files to repository
client.upload_file(
repository="username/my-repo",
local_path="./optimizer.py",
remote_path="optimizers/genetic_algorithm.py"
)
# Download files from repository
client.download_file(
repository="username/my-repo",
remote_path="problems/tsp_problem.py",
local_path="./downloaded_problem.py"
)
# List repository contents
files = client.list_repository_files("username/my-repo")
for file in files:
print(f"{file.path} ({file.size} bytes)")
Execution Services
Access cloud execution services:
# Submit optimization job
job_id = client.submit_optimization_job(
problem_repo="community/tsp_problem",
optimizer_repo="algorithms/genetic_tsp",
problem_params={"n_cities": 100},
optimizer_params={"population_size": 200, "generations": 1000},
resources={"cpu_cores": 4, "memory_gb": 8}
)
# Monitor job status
status = client.get_job_status(job_id)
print(f"Job status: {status.state}")
print(f"Progress: {status.progress:.1%}")
# Get results when complete
if status.state == "completed":
result = client.get_job_result(job_id)
print(f"Best value: {result.best_value}")
Benchmarking Services
Access platform benchmarking capabilities:
# Submit benchmark suite
benchmark_id = client.submit_benchmark_suite(
problems=["community/tsp_berlin52", "community/tsp_kroA100"],
optimizers=["algorithms/genetic_tsp", "algorithms/simulated_annealing"],
num_runs=30,
resources={"cpu_cores": 8, "memory_gb": 16}
)
# Monitor benchmark progress
progress = client.get_benchmark_progress(benchmark_id)
print(f"Completed: {progress.completed_runs}/{progress.total_runs}")
# Get benchmark results
results = client.get_benchmark_results(benchmark_id)
for result in results:
print(f"{result.problem} + {result.optimizer}: {result.mean_value:.4f}")
Advanced Features
Collaboration Features
Collaborate with other researchers:
# Follow researchers
client.follow_user("optimization_expert")
# Get notifications
notifications = client.get_notifications(category="new_models")
for notification in notifications:
print(f"{notification.title}: {notification.message}")
# Join research groups
client.join_group("quantum_optimization_research")
# Share models with groups
client.share_model_with_group("my_optimizer", "research_group")
Analytics and Insights
Access platform analytics:
# Get model usage statistics
stats = client.get_model_stats("my_optimizer")
print(f"Downloads: {stats.download_count}")
print(f"Executions: {stats.execution_count}")
print(f"Average rating: {stats.average_rating}")
# Get platform trends
trends = client.get_platform_trends(timeframe="month")
print(f"Popular problems: {trends.popular_problems}")
print(f"Popular optimizers: {trends.popular_optimizers}")
Custom Workflows
Create custom platform workflows:
# Define custom workflow
workflow = client.create_workflow("parameter_sweep")
# Add workflow steps
workflow.add_step("load_problem", {
"repository": "community/tsp_problem",
"params": {"n_cities": 50}
})
workflow.add_step("parameter_sweep", {
"optimizer_repository": "algorithms/genetic_tsp",
"parameter_ranges": {
"population_size": [50, 100, 200],
"mutation_rate": [0.01, 0.05, 0.1]
}
})
workflow.add_step("analyze_results", {
"metrics": ["best_value", "convergence_time", "success_rate"]
})
# Execute workflow
workflow_id = client.execute_workflow(workflow)
results = client.get_workflow_results(workflow_id)
Configuration Management
Client Configuration
Configure client behavior:
# Load configuration from file
config = client.load_config("~/.rastion/config.json")
# Update configuration
client.update_config({
"default_timeout": 3600,
"max_retries": 3,
"cache_enabled": True,
"cache_directory": "~/.rastion/cache"
})
# Save configuration
client.save_config("~/.rastion/config.json")
Environment Setup
Set up development environment:
# Initialize development environment
client.init_dev_environment(
workspace_path="./qubots-workspace",
template="optimization-research"
)
# Sync with platform
client.sync_workspace(
local_path="./qubots-workspace",
remote_repository="username/research-project"
)
Error Handling
Exception Types
Handle platform-specific errors:
from qubots.rastion_client import (
RastionAuthenticationError,
RastionAPIError,
RastionNetworkError,
RastionModelNotFoundError
)
try:
model = rastion.load_qubots_model("nonexistent_model", "username")
except RastionModelNotFoundError:
print("Model not found on platform")
except RastionAuthenticationError:
print("Authentication failed - check API token")
except RastionNetworkError:
print("Network error - check connection")
except RastionAPIError as e:
print(f"API error: {e.message}")
Retry Logic
Implement robust retry mechanisms:
import time
from typing import Optional
def robust_model_load(model_name: str, username: str, max_retries: int = 3):
"""Load model with retry logic."""
for attempt in range(max_retries):
try:
return rastion.load_qubots_model(model_name, username)
except RastionNetworkError:
if attempt < max_retries - 1:
wait_time = 2 ** attempt
print(f"Network error, retrying in {wait_time}s...")
time.sleep(wait_time)
else:
raise
Best Practices
Security
- Token Management: Store API tokens securely, never in code
- Permission Scoping: Use tokens with minimal required permissions
- Regular Rotation: Rotate API tokens regularly
- Environment Variables: Use environment variables for sensitive data
- Caching: Enable caching for frequently accessed models
- Batch Operations: Use batch APIs for multiple operations
- Connection Pooling: Reuse client instances when possible
- Timeout Configuration: Set appropriate timeouts for operations
Development Workflow
- Local Testing: Test models locally before uploading
- Version Control: Use semantic versioning for model releases
- Documentation: Provide comprehensive model documentation
- Testing: Include tests with model uploads
Next Steps