sangfor-scp

📖 Live Examples & API Docs →

Sangfor Cloud Platform (SCP) Open-API Python Client Library

A clean, Pythonic wrapper for the Sangfor Cloud Platform REST API. Supports EC2 (AWS4-HMAC-SHA256) and Token-based authentication with a single login on initialization — no repeated auth delays between calls.

Features

• Single login — authenticate once in __init__, session reused for all requests
• Auto token refresh — 24-hour tokens renewed transparently
• Automatic pagination — list_all() iterates every page lazily; no manual page tracking needed
• Async task support — tasks.wait() polls until completion, raises on failure or timeout
• Idempotency — automatic X-Client-Token headers on POST/PUT requests
• Both auth methods — EC2 signature (recommended) and Token-based (RSA encrypted)
• SSL flexibility — optional SSL verification disable (for self-signed certs)

Covered APIs (Top 20 Critical Features)

Installation

pip install sangfor-scp

With Token-based authentication support (RSA encryption):

pip install sangfor-scp[token-auth]

Quick Start

EC2 Authentication (Recommended)

from sangfor_scp import SCPClient

client = SCPClient(
    host="10.134.37.79",
    access_key="your_access_key",
    secret_key="your_secret_key",
    region="cn-south-1",
    verify_ssl=False,   # SCP uses self-signed certs by default
)

Token-Based Authentication

from sangfor_scp import SCPClient

client = SCPClient(
    host="10.134.37.79",
    username="admin",
    password="your_password",
    verify_ssl=False,
)

Usage Examples

List All Virtual Machines

# Iterates all pages automatically — no pagination code needed
for vm in client.servers.list_all():
    print(vm["id"], vm["name"], vm["status"])

# Filter by resource pool
for vm in client.servers.list_all(az_id="az-uuid", status="running"):
    print(vm["name"])

Create a VM and Wait for Completion

result = client.servers.create(
    az_id="az-uuid",
    image_id="image-uuid",
    storage_tag_id="storage-tag-uuid",
    cores=2,
    memory_mb=2048,
    name="my-vm",
    networks=[{
        "vif_id": "net0",
        "vpc_id": "vpc-uuid",
        "subnet_id": "subnet-uuid",
        "connect": 1,
        "model": "virtio",
    }],
    power_on=True,
)

# Wait up to 5 minutes for the VM to be ready
task = client.tasks.wait(result["task_id"], timeout=300)
vm_id = result["uuids"][0]
print(f"VM created: {vm_id}")

Rename / Update a VM

# Rename
task_id = client.servers.rename("vm-uuid", "new-name")
client.tasks.wait(task_id)

# Update multiple properties at once (only passed fields are changed)
task_id = client.servers.update(
    "vm-uuid",
    name="renamed-vm",
    cores=4,
    memory_mb=8192,
    description="Updated description",
)
client.tasks.wait(task_id)

Power Operations

client.servers.stop("vm-uuid")
client.servers.stop("vm-uuid", force=True)     # force power off
client.servers.start("vm-uuid")
client.servers.reboot("vm-uuid")
client.servers.reboot("vm-uuid", force=True)   # force reboot
client.servers.suspend("vm-uuid")

Clone a VM

result = client.servers.clone(
    "vm-uuid",
    name="cloned-vm",
    az_id="az-uuid",           # optional: target resource pool
    storage_tag_id="tag-uuid", # optional: target storage
)
client.tasks.wait(result["task_id"])
new_vm_id = result["uuid"]

Migrate a VM

# Migrate to a specific host (None = auto-select)
task_id = client.servers.migrate("vm-uuid", host_id="host-uuid")
client.tasks.wait(task_id)

Get Console URL (VNC/SPICE/RDP)

info = client.servers.get_console("vm-uuid")
print(info["url"])   # Open in browser for noVNC console

# SPICE console
info = client.servers.get_console("vm-uuid", protocol="spice")

Batch Operations

vm_ids = ["vm-uuid-1", "vm-uuid-2", "vm-uuid-3"]

# Soft delete (move to recycle bin)
client.servers.soft_delete(vm_ids)

# Restore from recycle bin
task_id = client.servers.restore("vm-uuid-1")
client.tasks.wait(task_id)

# Other batch actions
client.servers.batch_action(vm_ids, "start_servers_action")
client.servers.batch_action(vm_ids, "stop_servers_action")
client.servers.batch_action(vm_ids, "reboot_servers_action")

Disk Management

# Create a standalone disk
result = client.volumes.create(
    az_id="az-uuid",
    storage_tag_id="tag-uuid",
    size_mb=102400,   # 100 GB
    name="data-disk",
)
client.tasks.wait(result["task_id"])
volume_id = result["volume_id"]

# Attach to a VM
task_id = client.servers.attach_volume("vm-uuid", volume_id)
client.tasks.wait(task_id)

# Detach
task_id = client.servers.detach_volume("vm-uuid", volume_id)
client.tasks.wait(task_id)

Network Management

# List VPCs
for vpc in client.networks.list_vpcs(az_id="az-uuid"):
    print(vpc["id"], vpc["name"])

# Create Subnet
result = client.networks.create_subnet(
    vpc_id="vpc-uuid",
    az_id="az-uuid",
    cidr="192.168.10.0/24",
    name="subnet-app",
    gateway_ip="192.168.10.1",
)

# List subnets in a VPC
for subnet in client.networks.list_subnets(vpc_id="vpc-uuid"):
    print(subnet["id"], subnet["cidr"])

Elastic IP

# Allocate
eip = client.eips.allocate(az_id="az-uuid", bandwidth_mb=100)
eip_id = eip["id"]

# Bind to a VM
task_id = client.eips.bind(eip_id, server_id="vm-uuid")
client.tasks.wait(task_id)

# Unbind and release
client.eips.unbind(eip_id)
client.eips.release(eip_id)

Platform Overview

# Resource utilization summary
overview = client.resource_pools.overview()
print(f"Total VMs: {overview['server']['total']}")
print(f"Running VMs: {overview['server']['running_count']}")

# SCP version
info = client.system.version()
print(info["build_version"])

Async Task with Progress Callback

def on_progress(task_data):
    print(f"Progress: {task_data['progress']}% — {task_data.get('description', '')}")

task = client.tasks.wait(
    task_id="task-uuid",
    timeout=600,
    poll_interval=5,
    progress_callback=on_progress,
)

Manual Pagination

# Get a single page with full metadata
page = client.servers.list_page(page_num=0, page_size=10)
print(f"Total VMs: {page['total_size']}")
print(f"Next page: {page['next_page_num']}")

# Count without fetching all records
total = client.servers.count(az_id="az-uuid")
print(f"VM count in pool: {total}")

Error Handling

from sangfor_scp import (
    SCPError,
    SCPNotFoundError,
    SCPAuthError,
    SCPRateLimitError,
    SCPTaskError,
    SCPTimeoutError,
)

try:
    vm = client.servers.get("nonexistent-id")
except SCPNotFoundError:
    print("VM not found")

try:
    task = client.tasks.wait("task-uuid", timeout=60)
except SCPTaskError as e:
    print(f"Task failed: {e.message}")
    print(f"Task data: {e.task_data}")
except SCPTimeoutError as e:
    print(f"Timed out after {e.timeout}s")

Requirements

• Python >= 3.8
• requests >= 2.28.0
• urllib3 >= 1.26.0
• pycryptodome >= 3.17 (only for Token-based auth)

SCP Version Compatibility

License

MIT License — see LICENSE for details.

Special Thanks