or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

docs

authentication.mdcdn-management.mdcloud-computing.mdcommunication-services.mdconfiguration-utilities.mddata-processing.mdfile-storage.mdindex.md
tile.json

tessl/pypi-qiniu

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

Workspace
tessl
Visibility
Public
Created
Last updated
Describes
pypipkg:pypi/qiniu@7.17.x

To install, run

npx @tessl/cli install tessl/pypi-qiniu@7.17.0

index.mddocs/

Qiniu Python SDK

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): ...