tessl/pypi-qiniu

Comprehensive Python SDK for Qiniu Cloud Storage services enabling file upload, download, CDN management, SMS, and real-time communication features

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Securityby

Pending

The risk profile of this skill

Overview

Eval results

Files

Qiniu Python SDK

Name: tessl/pypi-qiniu
Author: tessl

A comprehensive Python SDK for Qiniu Cloud Storage services, enabling developers to integrate file upload, download, and management capabilities with Qiniu's cloud storage platform. The SDK offers a full-featured API for storage operations including bucket management, file operations, CDN management, SMS services, and real-time communication features.

Package Information

Package Name: qiniu
Language: Python
Installation: pip install qiniu
Version: 7.17.0

Core Imports

import qiniu

Most common usage patterns:

from qiniu import Auth, BucketManager, put_file_v2, put_data

Basic Usage

from qiniu import Auth, put_file_v2, BucketManager

# Initialize authentication
access_key = 'your_access_key'
secret_key = 'your_secret_key'
auth = Auth(access_key, secret_key)

# Upload a file
bucket_name = 'your_bucket'
key = 'file_key'
local_file = './path/to/file.jpg'

# Generate upload token
token = auth.upload_token(bucket_name, key, 3600)

# Upload file
ret, info = put_file_v2(token, key, local_file)
print(info)
print(ret['key'], ret['hash'])

# Manage files
bucket_manager = BucketManager(auth)
ret, info = bucket_manager.stat(bucket_name, key)
print(ret)

Architecture

The Qiniu SDK is organized around several key components:

Authentication: Auth and QiniuMacAuth classes provide secure token-based authentication
Storage Operations: BucketManager handles file and bucket management, upload functions handle file transfers
CDN Management: CdnManager and DomainManager control content delivery and domain configuration
Processing Services: PersistentFop enables data processing workflows
Compute Services: AccountClient and QcosClient manage cloud computing resources
Communication Services: Sms and RtcServer provide messaging and real-time communication
Utilities: Helper functions for encoding, hashing, and data transformation

Capabilities

Authentication & Security

Secure authentication using access/secret keys with token-based access control for all Qiniu services.

class Auth:
    def __init__(self, access_key: str, secret_key: str, disable_qiniu_timestamp_signature: bool = None): ...
    def upload_token(self, bucket: str, key: str = None, expires: int = 3600, policy: dict = None, strict_policy: bool = True) -> str: ...
    def private_download_url(self, url: str, expires: int = 3600) -> str: ...
    def token_of_request(self, url: str, body: str = None, content_type: str = None) -> str: ...

class QiniuMacAuth:
    def __init__(self, access_key: str, secret_key: str, disable_qiniu_timestamp_signature: bool = None): ...
    def token_of_request(self, method: str, host: str, url: str, qheaders: dict, content_type: str = None, body: str = None) -> str: ...

Authentication

File Storage Operations

Complete file upload, download, and management operations with support for resumable uploads and batch operations.

def put_file_v2(up_token: str, key: str, file_path: str, **kwargs) -> tuple: ...
def put_data(up_token: str, key: str, data: bytes, **kwargs) -> tuple: ...
def put_stream_v2(up_token: str, key: str, input_stream, file_name: str, data_size: int, **kwargs) -> tuple: ...

class BucketManager:
    def __init__(self, auth: Auth, **kwargs): ...
    def list(self, bucket: str, prefix: str = None, marker: str = None, limit: int = None, delimiter: str = None) -> tuple: ...
    def stat(self, bucket: str, key: str) -> tuple: ...
    def delete(self, bucket: str, key: str) -> tuple: ...
    def copy(self, bucket: str, key: str, bucket_to: str, key_to: str, force: str = 'false') -> tuple: ...
    def batch(self, operations: list) -> tuple: ...

File Storage

CDN Management

Content delivery network cache management, domain configuration, and traffic statistics.

class CdnManager:
    def __init__(self, auth: Auth): ...
    def refresh_urls(self, urls: list) -> tuple: ...
    def refresh_dirs(self, dirs: list) -> tuple: ...
    def prefetch_urls(self, urls: list) -> tuple: ...
    def get_bandwidth_data(self, domains: list, start_date: str, end_date: str, granularity: str, data_type: str = None) -> tuple: ...

class DomainManager:
    def __init__(self, auth: Auth): ...
    def create_domain(self, name: str, body: dict) -> tuple: ...
    def put_httpsconf(self, name: str, certid: str, forceHttps: bool) -> tuple: ...

CDN Management

Data Processing

Persistent data processing operations for media files including audio, video, and image transformations.

class PersistentFop:
    def __init__(self, auth: Auth, bucket: str, pipeline: str = None, notify_url: str = None): ...
    def execute(self, key: str, fops: str = None, force: bool = None, persistent_type: int = None, workflow_template_id: str = None) -> tuple: ...
    def get_status(self, persistent_id: str) -> tuple: ...

def build_op(cmd: str, first_arg: str, **kwargs) -> str: ...
def pipe_cmd(*cmds: str) -> str: ...
def op_save(op: str, bucket: str, key: str) -> str: ...

Data Processing

Cloud Computing

Application and service stack management for cloud computing resources including containers and access points.

class AccountClient:
    def __init__(self, auth: Auth, host: str = None): ...
    def list_apps(self) -> tuple: ...
    def create_app(self, args: dict) -> tuple: ...
    def get_qcos_client(self, app_uri: str): ...

class QcosClient:
    def __init__(self, auth: Auth, host: str = None): ...
    def list_stacks(self) -> tuple: ...
    def create_stack(self, args: dict) -> tuple: ...
    def list_services(self, stack: str) -> tuple: ...
    def list_containers(self, stack: str = None, service: str = None) -> tuple: ...

Cloud Computing

Communication Services

SMS messaging and real-time communication services for application integration.

class Sms:
    def __init__(self, auth: Auth): ...
    def createSignature(self, signature: str, source: str, pics: list = None) -> tuple: ...
    def createTemplate(self, name: str, template: str, type: str, description: str, signature_id: str) -> tuple: ...
    def sendMessage(self, template_id: str, mobiles: list, parameters: dict) -> tuple: ...

class RtcServer:
    def __init__(self, auth: Auth): ...
    def create_app(self, data: dict) -> tuple: ...
    def list_user(self, app_id: str, room_name: str) -> tuple: ...
    def kick_user(self, app_id: str, room_name: str, user_id: str) -> tuple: ...

def get_room_token(access_key: str, secret_key: str, room_access: dict) -> str: ...

Communication Services

Configuration & Utilities

Global configuration management, region/zone settings, and utility functions for encoding and data transformation.

def set_default(**kwargs): ...
def get_default(key: str): ...

class Region:
    def __init__(self, **kwargs): ...
    def get_up_host_by_token(self, up_token: str, home_dir: str) -> str: ...
    def get_bucket_hosts(self, ak: str, bucket: str, home_dir: str = None, force: bool = False) -> dict: ...

class Zone:
    """Zone configuration (legacy alias for Region)"""
    def __init__(self, **kwargs): ...
    def get_up_host_by_token(self, up_token: str, home_dir: str) -> str: ...
    def get_bucket_hosts(self, ak: str, bucket: str, home_dir: str = None, force: bool = False) -> dict: ...

def urlsafe_base64_encode(data: bytes) -> str: ...
def urlsafe_base64_decode(data: str) -> bytes: ...
def entry(bucket: str, key: str) -> str: ...
def etag(file_path: str) -> str: ...
def crc32(data: bytes) -> int: ...

Configuration & Utilities

Common Types

class ResponseInfo:
    """HTTP response information wrapper"""
    status_code: int
    text_body: str
    req_id: str
    x_log: str
    error: str
    url: str
    exception: Exception
    
    def ok(self) -> bool: ...
    def need_retry(self) -> bool: ...
    def connect_failed(self) -> bool: ...
    def json(self) -> dict: ...

class UploadProgressRecorder:
    """Persistent upload progress recording for resumable uploads"""
    def __init__(self, record_folder: str = None): ...
    def has_upload_record(self, file_name: str, key: str) -> bool: ...
    def get_upload_record(self, file_name: str, key: str) -> dict: ...
    def set_upload_record(self, file_name: str, key: str, data: dict): ...
    def delete_upload_record(self, file_name: str, key: str): ...