Comprehensive Python SDK for Qiniu Cloud Storage services enabling file upload, download, CDN management, SMS, and real-time communication features
—
Secure authentication using access/secret keys with token-based access control for all Qiniu services. The authentication system supports both standard Auth and MAC (Message Authentication Code) authentication methods.
The Auth class provides the primary authentication interface for Qiniu services.
class Auth:
def __init__(self, access_key: str, secret_key: str, disable_qiniu_timestamp_signature: bool = None):
"""
Initialize authentication with access and secret keys.
Args:
access_key: Qiniu access key
secret_key: Qiniu secret key
disable_qiniu_timestamp_signature: Whether to disable timestamp signatures
"""
def get_access_key(self) -> str:
"""Get the access key."""
def get_secret_key(self) -> str:
"""Get the secret key."""
def upload_token(self, bucket: str, key: str = None, expires: int = 3600, policy: dict = None, strict_policy: bool = True) -> str:
"""
Generate upload token for file uploads.
Args:
bucket: Target bucket name
key: File key (optional for auto-generated key)
expires: Token expiration time in seconds (default: 3600)
policy: Upload policy dictionary (optional)
strict_policy: Whether to enforce strict policy validation
Returns:
Upload token string
"""
def private_download_url(self, url: str, expires: int = 3600) -> str:
"""
Generate private download URL with expiration.
Args:
url: Original file URL
expires: URL expiration time in seconds
Returns:
Signed private download URL
"""
def token_of_request(self, url: str, body: str = None, content_type: str = None) -> str:
"""
Generate management token for API requests.
Args:
url: Request URL
body: Request body (optional)
content_type: Content type header (optional)
Returns:
Management token string
"""
def token_with_data(self, data: str) -> str:
"""
Generate token with encoded data.
Args:
data: Data to encode in token
Returns:
Token string with encoded data
"""
def verify_callback(self, origin_authorization: str, url: str, body: str, content_type: str = 'application/x-www-form-urlencoded', method: str = 'GET', headers: dict = None) -> bool:
"""
Verify callback request authorization.
Args:
origin_authorization: Original authorization header
url: Callback URL
body: Request body
content_type: Content type
method: HTTP method
headers: Request headers
Returns:
True if verification succeeds
"""
@staticmethod
def up_token_decode(up_token: str) -> tuple:
"""
Decode upload token to extract policy information.
Args:
up_token: Upload token to decode
Returns:
(access_key, signature, policy_dict): Tuple containing access key, signature, and policy data
"""
@staticmethod
def get_bucket_name(up_token: str) -> str:
"""
Extract bucket name from upload token.
Args:
up_token: Upload token
Returns:
Bucket name string
"""Advanced MAC authentication for direct HTTP client integration.
class QiniuMacAuth:
def __init__(self, access_key: str, secret_key: str, disable_qiniu_timestamp_signature: bool = None):
"""
Initialize MAC authentication.
Args:
access_key: Qiniu access key
secret_key: Qiniu secret key
disable_qiniu_timestamp_signature: Whether to disable timestamp signatures
"""
def token_of_request(self, method: str, host: str, url: str, qheaders: dict, content_type: str = None, body: str = None) -> str:
"""
Generate request token using MAC authentication.
Args:
method: HTTP method (GET, POST, etc.)
host: Request host
url: Request URL path
qheaders: Qiniu-specific headers
content_type: Request content type
body: Request body
Returns:
MAC authentication token
"""
def qiniu_headers(self, headers: dict) -> dict:
"""
Process and filter Qiniu-specific headers.
Args:
headers: Original headers dictionary
Returns:
Filtered Qiniu headers
"""
def verify_callback(self, origin_authorization: str, url: str, body: str, content_type: str, method: str, headers: dict) -> bool:
"""
Verify callback authorization using MAC.
Args:
origin_authorization: Original authorization header
url: Callback URL
body: Request body
content_type: Content type
method: HTTP method
headers: Request headers
Returns:
True if verification succeeds
"""
@property
def should_sign_with_timestamp(self) -> bool:
"""Whether timestamp signature should be used."""Authentication adapters for use with the requests library.
class RequestsAuth:
def __init__(self, auth: Auth):
"""
Requests library auth adapter for basic Auth.
Args:
auth: Auth instance
"""
class QiniuMacRequestsAuth:
def __init__(self, auth: QiniuMacAuth):
"""
Requests library auth adapter for MAC authentication.
Args:
auth: QiniuMacAuth instance
"""from qiniu import Auth
# Initialize with your credentials
access_key = 'your_access_key'
secret_key = 'your_secret_key'
auth = Auth(access_key, secret_key)
# Generate upload token
bucket = 'my-bucket'
key = 'my-file.jpg'
token = auth.upload_token(bucket, key, expires=7200) # 2 hours
print(f"Upload token: {token}")from qiniu import Auth
auth = Auth(access_key, secret_key)
# Custom upload policy
policy = {
'scope': 'my-bucket:my-file.jpg',
'deadline': 1893456000, # Unix timestamp
'insertOnly': 1, # Prevent overwrite
'fsizeLimit': 10485760, # Max 10MB
'callbackUrl': 'https://api.example.com/callback',
'callbackBody': 'filename=$(fname)&filesize=$(fsize)'
}
token = auth.upload_token('my-bucket', 'my-file.jpg', policy=policy)from qiniu import Auth
auth = Auth(access_key, secret_key)
# Original public URL
public_url = 'http://domain.com/file.jpg'
# Generate private download URL valid for 1 hour
private_url = auth.private_download_url(public_url, expires=3600)
print(f"Private URL: {private_url}")import requests
from qiniu import QiniuMacAuth, QiniuMacRequestsAuth
mac_auth = QiniuMacAuth(access_key, secret_key)
requests_auth = QiniuMacRequestsAuth(mac_auth)
# Use with requests library
response = requests.get('https://rs.qiniu.com/bucket/my-bucket', auth=requests_auth)
print(response.json())Install with Tessl CLI
npx tessl i tessl/pypi-qiniu