CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-qiniu

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

—

Pending
Overview
Eval results
Files

cdn-management.mddocs/

CDN Management

Content delivery network cache management, domain configuration, and traffic statistics for optimizing content delivery performance and managing CDN-related operations.

Capabilities

CDN Cache Management

Cache refresh, prefetch operations, and traffic statistics through the CdnManager class.

class CdnManager:
    def __init__(self, auth: Auth):
        """
        Initialize CDN manager.
        
        Args:
            auth: Auth instance for authentication
        """

    def refresh_urls(self, urls: list) -> tuple:
        """
        Refresh cache for specific URLs.
        
        Args:
            urls: List of URLs to refresh
            
        Returns:
            (dict, ResponseInfo): Refresh result and response info
        """

    def refresh_dirs(self, dirs: list) -> tuple:
        """
        Refresh cache for directories.
        
        Args:
            dirs: List of directory URLs to refresh
            
        Returns:
            (dict, ResponseInfo): Refresh result and response info
        """

    def refresh_urls_and_dirs(self, urls: list, dirs: list) -> tuple:
        """
        Refresh cache for both URLs and directories.
        
        Args:
            urls: List of URLs to refresh
            dirs: List of directory URLs to refresh
            
        Returns:
            (dict, ResponseInfo): Refresh result and response info
        """

    def prefetch_urls(self, urls: list) -> tuple:
        """
        Prefetch files to CDN edge nodes.
        
        Args:
            urls: List of URLs to prefetch
            
        Returns:
            (dict, ResponseInfo): Prefetch result and response info
        """

    def get_bandwidth_data(self, domains: list, start_date: str, end_date: str, granularity: str, data_type: str = None) -> tuple:
        """
        Get bandwidth statistics.
        
        Args:
            domains: List of domain names
            start_date: Start date (YYYY-MM-DD format)
            end_date: End date (YYYY-MM-DD format)
            granularity: Data granularity ('5min', 'hour', 'day')
            data_type: Data type from DataType enum (optional)
            
        Returns:
            (dict, ResponseInfo): Bandwidth data and response info
        """

    def get_flux_data(self, domains: list, start_date: str, end_date: str, granularity: str, data_type: str = None) -> tuple:
        """
        Get traffic statistics.
        
        Args:
            domains: List of domain names
            start_date: Start date (YYYY-MM-DD format)
            end_date: End date (YYYY-MM-DD format)
            granularity: Data granularity ('5min', 'hour', 'day')
            data_type: Data type from DataType enum (optional)
            
        Returns:
            (dict, ResponseInfo): Traffic data and response info
        """

    def get_log_list_data(self, domains: list, log_date: str) -> tuple:
        """
        Get CDN log download links.
        
        Args:
            domains: List of domain names
            log_date: Log date (YYYY-MM-DD format)
            
        Returns:
            (dict, ResponseInfo): Log download links and response info
        """

    def put_httpsconf(self, name: str, certid: str, forceHttps: bool = False) -> tuple:
        """
        Update HTTPS certificate configuration.
        
        Args:
            name: Domain name
            certid: Certificate ID
            forceHttps: Force HTTPS redirect
            
        Returns:
            (dict, ResponseInfo): Configuration result and response info
        """

Domain Management

Advanced domain configuration and SSL certificate management through the DomainManager class.

class DomainManager:
    def __init__(self, auth: Auth):
        """
        Initialize domain manager.
        
        Args:
            auth: Auth instance for authentication
        """

    def create_domain(self, name: str, body: dict) -> tuple:
        """
        Create new domain configuration.
        
        Args:
            name: Domain name
            body: Domain configuration dictionary
            
        Returns:
            (dict, ResponseInfo): Creation result and response info
        """

    def domain_online(self, name: str) -> tuple:
        """
        Bring domain online.
        
        Args:
            name: Domain name
            
        Returns:
            (dict, ResponseInfo): Operation result and response info
        """

    def domain_offline(self, name: str) -> tuple:
        """
        Take domain offline.
        
        Args:
            name: Domain name
            
        Returns:
            (dict, ResponseInfo): Operation result and response info
        """

    def delete_domain(self, name: str) -> tuple:
        """
        Delete domain configuration.
        
        Args:
            name: Domain name
            
        Returns:
            (dict, ResponseInfo): Deletion result and response info
        """

    def get_domain(self, name: str) -> tuple:
        """
        Get domain configuration information.
        
        Args:
            name: Domain name
            
        Returns:
            (dict, ResponseInfo): Domain info and response info
        """

    def put_httpsconf(self, name: str, certid: str, forceHttps: bool) -> tuple:
        """
        Update HTTPS configuration.
        
        Args:
            name: Domain name
            certid: Certificate ID
            forceHttps: Force HTTPS redirect
            
        Returns:
            (dict, ResponseInfo): Configuration result and response info
        """

    def create_sslcert(self, name: str, common_name: str, pri: str, ca: str) -> tuple:
        """
        Create SSL certificate.
        
        Args:
            name: Certificate name
            common_name: Certificate common name
            pri: Private key content
            ca: Certificate authority content
            
        Returns:
            (dict, ResponseInfo): Certificate creation result and response info
        """

CDN Data Types

Constants for specifying data types in statistics queries.

class DataType:
    """CDN data type constants for statistics queries."""
    BANDWIDTH = 'bandwidth'
    X302BANDWIDTH = '302bandwidth'
    X302MBANDWIDTH = '302mbandwidth'
    FLOW = 'flow'
    X302FLOW = '302flow'
    X302MFLOW = '302mflow'

Anti-Leech URL Generation

Utility function for creating secure URLs with timestamp-based access control.

def create_timestamp_anti_leech_url(host: str, file_name: str, query_string: str, encrypt_key: str, deadline: int) -> str:
    """
    Create timestamp-based anti-leech URL.
    
    Args:
        host: Domain host
        file_name: File path
        query_string: URL query string
        encrypt_key: Encryption key for signature
        deadline: URL expiration timestamp
        
    Returns:
        Secure URL with anti-leech protection
    """

Usage Examples

Cache Management

from qiniu import Auth, CdnManager

auth = Auth(access_key, secret_key)
cdn_manager = CdnManager(auth)

# Refresh specific URLs
urls_to_refresh = [
    'https://cdn.example.com/image1.jpg',
    'https://cdn.example.com/image2.png'
]
ret, info = cdn_manager.refresh_urls(urls_to_refresh)

if info.ok():
    print(f"Refresh request ID: {ret['requestId']}")
else:
    print(f"Refresh failed: {info.error}")

# Refresh directories
dirs_to_refresh = [
    'https://cdn.example.com/images/',
    'https://cdn.example.com/videos/'
]
ret, info = cdn_manager.refresh_dirs(dirs_to_refresh)

# Prefetch popular content
popular_urls = [
    'https://cdn.example.com/popular-video.mp4',
    'https://cdn.example.com/trending-image.jpg'
]
ret, info = cdn_manager.prefetch_urls(popular_urls)

Traffic Statistics

from qiniu import Auth, CdnManager, DataType

auth = Auth(access_key, secret_key)
cdn_manager = CdnManager(auth)

# Get bandwidth data for last 7 days
domains = ['cdn.example.com', 'static.example.com']
ret, info = cdn_manager.get_bandwidth_data(
    domains=domains,
    start_date='2024-01-01',
    end_date='2024-01-07',
    granularity='day',
    data_type=DataType.BANDWIDTH
)

if info.ok():
    for time_point in ret['data']:
        timestamp = time_point['time']
        bandwidth = time_point['values']['bandwidth']
        print(f"Time: {timestamp}, Bandwidth: {bandwidth} bps")

# Get traffic data
ret, info = cdn_manager.get_flux_data(
    domains=domains,
    start_date='2024-01-01',
    end_date='2024-01-07',
    granularity='hour',
    data_type=DataType.FLOW
)

if info.ok():
    total_traffic = sum(point['values']['flow'] for point in ret['data'])
    print(f"Total traffic: {total_traffic} bytes")

Domain Management

from qiniu import Auth, DomainManager

auth = Auth(access_key, secret_key)
domain_manager = DomainManager(auth)

# Create new domain
domain_config = {
    'name': 'cdn.example.com',
    'type': 'normal',
    'sources': [
        {
            'source': 'origin.example.com',
            'host': 'origin.example.com'
        }
    ]
}

ret, info = domain_manager.create_domain('cdn.example.com', domain_config)

if info.ok():
    print("Domain created successfully")

# Get domain information
ret, info = domain_manager.get_domain('cdn.example.com')
if info.ok():
    print(f"Domain status: {ret['operatingState']}")
    print(f"CNAME: {ret['cname']}")

# Configure HTTPS
ret, info = domain_manager.put_httpsconf(
    name='cdn.example.com',
    certid='cert-id-123',
    forceHttps=True
)

# Take domain offline for maintenance
ret, info = domain_manager.domain_offline('cdn.example.com')
print(f"Domain offline result: {info.ok()}")

# Bring domain back online
ret, info = domain_manager.domain_online('cdn.example.com')
print(f"Domain online result: {info.ok()}")

SSL Certificate Management

from qiniu import Auth, DomainManager

auth = Auth(access_key, secret_key)
domain_manager = DomainManager(auth)

# Create SSL certificate
with open('private.key', 'r') as f:
    private_key = f.read()

with open('certificate.crt', 'r') as f:
    certificate = f.read()

ret, info = domain_manager.create_sslcert(
    name='my-ssl-cert',
    common_name='cdn.example.com',
    pri=private_key,
    ca=certificate
)

if info.ok():
    cert_id = ret['certid']
    print(f"SSL certificate created: {cert_id}")
    
    # Apply certificate to domain
    ret, info = domain_manager.put_httpsconf(
        name='cdn.example.com',
        certid=cert_id,
        forceHttps=True
    )

Access Log Analysis

from qiniu import Auth, CdnManager
import requests

auth = Auth(access_key, secret_key)
cdn_manager = CdnManager(auth)

# Get log download links
ret, info = cdn_manager.get_log_list_data(
    domains=['cdn.example.com'],
    log_date='2024-01-15'
)

if info.ok():
    for domain_logs in ret['data']:
        domain = domain_logs['domain']
        for log_info in domain_logs['logs']:
            log_url = log_info['url']
            log_size = log_info['size']
            
            print(f"Domain: {domain}")
            print(f"Log URL: {log_url}")
            print(f"Log size: {log_size} bytes")
            
            # Download and process log file
            response = requests.get(log_url)
            if response.status_code == 200:
                # Process log content
                log_content = response.text
                lines = log_content.split('\n')
                print(f"Log entries: {len(lines)}")

Anti-Leech Protection

from qiniu import create_timestamp_anti_leech_url
import time

# Create secure URL with expiration
host = 'cdn.example.com'
file_name = '/protected/video.mp4'
query_string = ''
encrypt_key = 'your-encrypt-key'
deadline = int(time.time()) + 3600  # Valid for 1 hour

secure_url = create_timestamp_anti_leech_url(
    host=host,
    file_name=file_name,
    query_string=query_string,
    encrypt_key=encrypt_key,
    deadline=deadline
)

print(f"Secure URL: https://{host}{secure_url}")

Install with Tessl CLI

npx tessl i tessl/pypi-qiniu

docs

authentication.md

cdn-management.md

cloud-computing.md

communication-services.md

configuration-utilities.md

data-processing.md

file-storage.md

index.md

tile.json