tessl/pypi-qiniu
Comprehensive Python SDK for Qiniu Cloud Storage services enabling file upload, download, CDN management, SMS, and real-time communication features
—
Pending
cloud-computing.mddocs/
Cloud Computing
Application and service stack management for cloud computing resources including containers, access points, and volume management. The compute platform provides comprehensive infrastructure management capabilities.
Capabilities
Account and Application Management
Account-level operations and application lifecycle management through the AccountClient class.
class AccountClient:
def __init__(self, auth: Auth, host: str = None):
"""
Initialize account client.
Args:
auth: Auth instance for authentication
host: Custom host URL (optional)
"""
def get_account_info(self) -> tuple:
"""
Get account information.
Returns:
(dict, ResponseInfo): Account info and response info
"""
def list_regions(self) -> tuple:
"""
List available regions.
Returns:
(dict, ResponseInfo): Region list and response info
"""
def list_apps(self) -> tuple:
"""
List applications.
Returns:
(dict, ResponseInfo): Application list and response info
"""
def create_app(self, args: dict) -> tuple:
"""
Create new application.
Args:
args: Application configuration dictionary
Returns:
(dict, ResponseInfo): Creation result and response info
"""
def delete_app(self, app_uri: str) -> tuple:
"""
Delete application.
Args:
app_uri: Application URI identifier
Returns:
(dict, ResponseInfo): Deletion result and response info
"""
def get_app_keys(self, app_uri: str) -> tuple:
"""
Get application keys.
Args:
app_uri: Application URI identifier
Returns:
(dict, ResponseInfo): Application keys and response info
"""
def get_valid_app_auth(self, app_uri: str) -> tuple:
"""
Get valid application authentication.
Args:
app_uri: Application URI identifier
Returns:
(dict, ResponseInfo): Authentication info and response info
"""
def get_app_region_products(self, app_uri: str) -> tuple:
"""
Get application region products.
Args:
app_uri: Application URI identifier
Returns:
(dict, ResponseInfo): Region products and response info
"""
def get_region_products(self, region: str) -> tuple:
"""
Get products available in region.
Args:
region: Region identifier
Returns:
(dict, ResponseInfo): Region products and response info
"""
def get_qcos_client(self, app_uri: str):
"""
Get cached resource client for application.
Args:
app_uri: Application URI identifier
Returns:
QcosClient: Resource management client
"""
def create_qcos_client(self, app_uri: str):
"""
Create new resource client for application.
Args:
app_uri: Application URI identifier
Returns:
QcosClient: Resource management client
"""Resource Management
Comprehensive resource management including stacks, services, containers, and access points through the QcosClient class.
class QcosClient:
def __init__(self, auth: Auth, host: str = None):
"""
Initialize resource management client.
Args:
auth: Auth instance for authentication
host: Custom host URL (optional)
"""
# Stack Operations
def list_stacks(self) -> tuple:
"""
List service stacks.
Returns:
(dict, ResponseInfo): Stack list and response info
"""
def create_stack(self, args: dict) -> tuple:
"""
Create service stack.
Args:
args: Stack configuration dictionary
Returns:
(dict, ResponseInfo): Creation result and response info
"""
def delete_stack(self, stack: str) -> tuple:
"""
Delete service stack.
Args:
stack: Stack identifier
Returns:
(dict, ResponseInfo): Deletion result and response info
"""
def get_stack(self, stack: str) -> tuple:
"""
Get stack information.
Args:
stack: Stack identifier
Returns:
(dict, ResponseInfo): Stack info and response info
"""
def start_stack(self, stack: str) -> tuple:
"""
Start service stack.
Args:
stack: Stack identifier
Returns:
(dict, ResponseInfo): Operation result and response info
"""
def stop_stack(self, stack: str) -> tuple:
"""
Stop service stack.
Args:
stack: Stack identifier
Returns:
(dict, ResponseInfo): Operation result and response info
"""
# Service Operations
def list_services(self, stack: str) -> tuple:
"""
List services in stack.
Args:
stack: Stack identifier
Returns:
(dict, ResponseInfo): Service list and response info
"""
def create_service(self, stack: str, args: dict) -> tuple:
"""
Create service in stack.
Args:
stack: Stack identifier
args: Service configuration dictionary
Returns:
(dict, ResponseInfo): Creation result and response info
"""
def delete_service(self, stack: str, service: str) -> tuple:
"""
Delete service from stack.
Args:
stack: Stack identifier
service: Service identifier
Returns:
(dict, ResponseInfo): Deletion result and response info
"""
def get_service_inspect(self, stack: str, service: str) -> tuple:
"""
Get detailed service information.
Args:
stack: Stack identifier
service: Service identifier
Returns:
(dict, ResponseInfo): Service details and response info
"""
def start_service(self, stack: str, service: str) -> tuple:
"""
Start service.
Args:
stack: Stack identifier
service: Service identifier
Returns:
(dict, ResponseInfo): Operation result and response info
"""
def stop_service(self, stack: str, service: str) -> tuple:
"""
Stop service.
Args:
stack: Stack identifier
service: Service identifier
Returns:
(dict, ResponseInfo): Operation result and response info
"""
def update_service(self, stack: str, service: str, args: dict) -> tuple:
"""
Update service configuration.
Args:
stack: Stack identifier
service: Service identifier
args: Updated configuration dictionary
Returns:
(dict, ResponseInfo): Update result and response info
"""
def scale_service(self, stack: str, service: str, args: dict) -> tuple:
"""
Scale service replicas.
Args:
stack: Stack identifier
service: Service identifier
args: Scaling configuration dictionary
Returns:
(dict, ResponseInfo): Scaling result and response info
"""
# Volume Operations
def create_service_volume(self, stack: str, service: str, args: dict) -> tuple:
"""
Create volume for service.
Args:
stack: Stack identifier
service: Service identifier
args: Volume configuration dictionary
Returns:
(dict, ResponseInfo): Creation result and response info
"""
def extend_service_volume(self, stack: str, service: str, volume: str, args: dict) -> tuple:
"""
Extend service volume capacity.
Args:
stack: Stack identifier
service: Service identifier
volume: Volume identifier
args: Extension configuration dictionary
Returns:
(dict, ResponseInfo): Extension result and response info
"""
def delete_service_volume(self, stack: str, service: str, volume: str) -> tuple:
"""
Delete service volume.
Args:
stack: Stack identifier
service: Service identifier
volume: Volume identifier
Returns:
(dict, ResponseInfo): Deletion result and response info
"""
# Container Operations
def list_containers(self, stack: str = None, service: str = None) -> tuple:
"""
List containers.
Args:
stack: Stack identifier (optional filter)
service: Service identifier (optional filter)
Returns:
(dict, ResponseInfo): Container list and response info
"""
def get_container_inspect(self, ip: str) -> tuple:
"""
Get detailed container information.
Args:
ip: Container IP address
Returns:
(dict, ResponseInfo): Container details and response info
"""
def start_container(self, ip: str) -> tuple:
"""
Start container.
Args:
ip: Container IP address
Returns:
(dict, ResponseInfo): Operation result and response info
"""
def stop_container(self, ip: str) -> tuple:
"""
Stop container.
Args:
ip: Container IP address
Returns:
(dict, ResponseInfo): Operation result and response info
"""
def restart_container(self, ip: str) -> tuple:
"""
Restart container.
Args:
ip: Container IP address
Returns:
(dict, ResponseInfo): Operation result and response info
"""
# Access Point Operations
def list_aps(self) -> tuple:
"""
List access points.
Returns:
(dict, ResponseInfo): Access point list and response info
"""
def create_ap(self, args: dict) -> tuple:
"""
Create access point.
Args:
args: Access point configuration dictionary
Returns:
(dict, ResponseInfo): Creation result and response info
"""
def search_ap(self, mode: str, query: str) -> tuple:
"""
Search access points.
Args:
mode: Search mode
query: Search query string
Returns:
(dict, ResponseInfo): Search results and response info
"""
def get_ap(self, apid: str) -> tuple:
"""
Get access point information.
Args:
apid: Access point ID
Returns:
(dict, ResponseInfo): Access point info and response info
"""
def update_ap(self, apid: str, args: dict) -> tuple:
"""
Update access point configuration.
Args:
apid: Access point ID
args: Updated configuration dictionary
Returns:
(dict, ResponseInfo): Update result and response info
"""
def delete_ap(self, apid: str) -> tuple:
"""
Delete access point.
Args:
apid: Access point ID
Returns:
(dict, ResponseInfo): Deletion result and response info
"""
def set_ap_port(self, apid: str, port: int, args: dict) -> tuple:
"""
Set access point port configuration.
Args:
apid: Access point ID
port: Port number
args: Port configuration dictionary
Returns:
(dict, ResponseInfo): Configuration result and response info
"""
def publish_ap(self, apid: str, args: dict) -> tuple:
"""
Bind custom domain to access point.
Args:
apid: Access point ID
args: Domain binding configuration
Returns:
(dict, ResponseInfo): Binding result and response info
"""
def unpublish_ap(self, apid: str, args: dict) -> tuple:
"""
Unbind custom domain from access point.
Args:
apid: Access point ID
args: Domain unbinding configuration
Returns:
(dict, ResponseInfo): Unbinding result and response info
"""
def get_ap_port_healthcheck(self, apid: str, port: int) -> tuple:
"""
Get access point port health check configuration.
Args:
apid: Access point ID
port: Port number
Returns:
(dict, ResponseInfo): Health check config and response info
"""
def set_ap_port_container(self, apid: str, port: int, args: dict) -> tuple:
"""
Set backend container for access point port.
Args:
apid: Access point ID
port: Port number
args: Container configuration dictionary
Returns:
(dict, ResponseInfo): Configuration result and response info
"""
def disable_ap_port(self, apid: str, port: int) -> tuple:
"""
Disable access point port.
Args:
apid: Access point ID
port: Port number
Returns:
(dict, ResponseInfo): Operation result and response info
"""
def enable_ap_port(self, apid: str, port: int) -> tuple:
"""
Enable access point port.
Args:
apid: Access point ID
port: Port number
Returns:
(dict, ResponseInfo): Operation result and response info
"""
def get_ap_providers(self) -> tuple:
"""
List access point providers.
Returns:
(dict, ResponseInfo): Provider list and response info
"""
def get_web_proxy(self, backend: str) -> tuple:
"""
Get web proxy configuration.
Args:
backend: Backend identifier
Returns:
(dict, ResponseInfo): Proxy config and response info
"""Usage Examples
Application Management
from qiniu import Auth, AccountClient
auth = Auth(access_key, secret_key)
account_client = AccountClient(auth)
# Get account information
ret, info = account_client.get_account_info()
if info.ok():
print(f"Account ID: {ret['id']}")
print(f"Account name: {ret['name']}")
# List available regions
ret, info = account_client.list_regions()
if info.ok():
for region in ret['items']:
print(f"Region: {region['id']} - {region['name']}")
# Create new application
app_config = {
'name': 'my-web-app',
'region': 'cn-east-1',
'description': 'Web application deployment'
}
ret, info = account_client.create_app(app_config)
if info.ok():
app_uri = ret['uri']
print(f"Application created: {app_uri}")
# Get application keys
ret, info = account_client.get_app_keys(app_uri)
if info.ok():
print(f"App access key: {ret['accessKey']}")
# List all applications
ret, info = account_client.list_apps()
if info.ok():
for app in ret['items']:
print(f"App: {app['name']} ({app['uri']})")Stack and Service Management
from qiniu import Auth, AccountClient
auth = Auth(access_key, secret_key)
account_client = AccountClient(auth)
# Get resource client for application
app_uri = 'qcos://my-web-app'
qcos_client = account_client.get_qcos_client(app_uri)
# Create service stack
stack_config = {
'name': 'web-stack',
'description': 'Web service stack',
'region': 'cn-east-1'
}
ret, info = qcos_client.create_stack(stack_config)
if info.ok():
stack_id = ret['name']
print(f"Stack created: {stack_id}")
# Create web service
service_config = {
'name': 'web-server',
'image': 'nginx:alpine',
'replicas': 2,
'resources': {
'cpu': '500m',
'memory': '512Mi'
},
'ports': [
{
'containerPort': 80,
'protocol': 'TCP'
}
],
'env': [
{
'name': 'ENV',
'value': 'production'
}
]
}
ret, info = qcos_client.create_service(stack_id, service_config)
if info.ok():
service_id = ret['name']
print(f"Service created: {service_id}")
# Get service details
ret, info = qcos_client.get_service_inspect(stack_id, service_id)
if info.ok():
print(f"Service status: {ret['status']}")
print(f"Replicas: {ret['replicas']}")
# List all stacks
ret, info = qcos_client.list_stacks()
if info.ok():
for stack in ret['items']:
print(f"Stack: {stack['name']} - Status: {stack['status']}")
# List services in stack
ret, info = qcos_client.list_services(stack['name'])
if info.ok():
for service in ret['items']:
print(f" Service: {service['name']} - {service['status']}")Container Management
from qiniu import Auth, AccountClient
auth = Auth(access_key, secret_key)
account_client = AccountClient(auth)
qcos_client = account_client.get_qcos_client('qcos://my-web-app')
# List all containers
ret, info = qcos_client.list_containers()
if info.ok():
for container in ret['items']:
container_ip = container['ip']
print(f"Container: {container['name']} - IP: {container_ip}")
print(f"Status: {container['status']}")
# Get detailed container information
ret, info = qcos_client.get_container_inspect(container_ip)
if info.ok():
container_details = ret
print(f"Image: {container_details['image']}")
print(f"CPU Usage: {container_details.get('cpuUsage', 'N/A')}")
print(f"Memory Usage: {container_details.get('memoryUsage', 'N/A')}")
# Container lifecycle operations
container_ip = '10.0.1.100'
# Stop container
ret, info = qcos_client.stop_container(container_ip)
if info.ok():
print("Container stopped successfully")
# Start container
ret, info = qcos_client.start_container(container_ip)
if info.ok():
print("Container started successfully")
# Restart container
ret, info = qcos_client.restart_container(container_ip)
if info.ok():
print("Container restarted successfully")Service Scaling and Updates
from qiniu import Auth, AccountClient
auth = Auth(access_key, secret_key)
account_client = AccountClient(auth)
qcos_client = account_client.get_qcos_client('qcos://my-web-app')
stack_id = 'web-stack'
service_id = 'web-server'
# Scale service replicas
scale_config = {
'replicas': 5
}
ret, info = qcos_client.scale_service(stack_id, service_id, scale_config)
if info.ok():
print("Service scaled successfully")
# Update service configuration
update_config = {
'image': 'nginx:1.21-alpine',
'resources': {
'cpu': '1000m',
'memory': '1Gi'
},
'env': [
{
'name': 'ENV',
'value': 'production'
},
{
'name': 'LOG_LEVEL',
'value': 'info'
}
]
}
ret, info = qcos_client.update_service(stack_id, service_id, update_config)
if info.ok():
print("Service updated successfully")Volume Management
from qiniu import Auth, AccountClient
auth = Auth(access_key, secret_key)
account_client = AccountClient(auth)
qcos_client = account_client.get_qcos_client('qcos://my-web-app')
stack_id = 'web-stack'
service_id = 'web-server'
# Create volume for service
volume_config = {
'name': 'web-data',
'size': '10Gi',
'mountPath': '/var/www/html',
'storageClass': 'ssd'
}
ret, info = qcos_client.create_service_volume(stack_id, service_id, volume_config)
if info.ok():
volume_id = ret['name']
print(f"Volume created: {volume_id}")
# Extend volume capacity
extend_config = {
'size': '20Gi'
}
ret, info = qcos_client.extend_service_volume(stack_id, service_id, volume_id, extend_config)
if info.ok():
print("Volume extended successfully")Access Point Configuration
from qiniu import Auth, AccountClient
auth = Auth(access_key, secret_key)
account_client = AccountClient(auth)
qcos_client = account_client.get_qcos_client('qcos://my-web-app')
# Create access point
ap_config = {
'name': 'web-access-point',
'type': 'public',
'region': 'cn-east-1'
}
ret, info = qcos_client.create_ap(ap_config)
if info.ok():
ap_id = ret['id']
print(f"Access point created: {ap_id}")
# Configure port routing
port_config = {
'targetPort': 80,
'protocol': 'HTTP',
'healthCheck': {
'path': '/health',
'intervalSeconds': 30
}
}
ret, info = qcos_client.set_ap_port(ap_id, 80, port_config)
if info.ok():
print("Port configuration set")
# Bind custom domain
domain_config = {
'domain': 'api.example.com',
'sslEnabled': True
}
ret, info = qcos_client.publish_ap(ap_id, domain_config)
if info.ok():
print("Custom domain bound successfully")
# List all access points
ret, info = qcos_client.list_aps()
if info.ok():
for ap in ret['items']:
print(f"Access Point: {ap['name']} - Status: {ap['status']}")
print(f"Public URL: {ap.get('publicUrl', 'N/A')}")Infrastructure Monitoring
from qiniu import Auth, AccountClient
import time
auth = Auth(access_key, secret_key)
account_client = AccountClient(auth)
qcos_client = account_client.get_qcos_client('qcos://my-web-app')
def monitor_infrastructure():
"""Monitor infrastructure health"""
while True:
print("\n=== Infrastructure Status ===")
# Check stacks
ret, info = qcos_client.list_stacks()
if info.ok():
for stack in ret['items']:
print(f"Stack: {stack['name']} - {stack['status']}")
# Check services in stack
ret, info = qcos_client.list_services(stack['name'])
if info.ok():
for service in ret['items']:
print(f" Service: {service['name']} - {service['status']}")
print(f" Replicas: {service.get('replicas', 0)}")
# Check containers
ret, info = qcos_client.list_containers()
if info.ok():
healthy_containers = sum(1 for c in ret['items'] if c['status'] == 'running')
total_containers = len(ret['items'])
print(f"Containers: {healthy_containers}/{total_containers} healthy")
# Check access points
ret, info = qcos_client.list_aps()
if info.ok():
active_aps = sum(1 for ap in ret['items'] if ap['status'] == 'active')
total_aps = len(ret['items'])
print(f"Access Points: {active_aps}/{total_aps} active")
time.sleep(60) # Check every minute
# Start monitoring (run in background)
import threading
monitor_thread = threading.Thread(target=monitor_infrastructure)
monitor_thread.daemon = True
monitor_thread.start()Install with Tessl CLI
npx tessl i tessl/pypi-qiniu