Python

Python SDK

Official Python SDK for Fluxum API

WebSocket-based
End-to-end encryption
Async support
Python 3.8+

Quick Start

Get up and running with the Fluxum Python SDK in minutes

Using pip

pip install fluxum-sdk-python

Using conda

conda install -c fluxum fluxum-sdk-python

Async Support

Full async/await support with Python's asyncio for high-performance applications.

Security

Built-in client-side encryption with optional custom encryption keys for maximum security.

Analytics

Comprehensive analytics tracking for connections, jobs, and rate limits with Python streams.

Event Handling

Handle real-time events with the SDK

@sdk.on(SDK_EVENTS.CONNECTED)
def on_connected():
    print('Connected to Fluxum server')

@sdk.on(SDK_EVENTS.AUTHENTICATED)
def on_authenticated(data):
    print(f'Authentication successful: {data.client_id}')

@sdk.on(SDK_EVENTS.DISCONNECTED)
def on_disconnected(data):
    print(f'Disconnected: {data.reason}')

@sdk.on(SDK_EVENTS.JOB_SUBMITTED)
def on_job_submitted(data):
    print(f'Job submitted: {data.job_id}')

@sdk.on(SDK_EVENTS.JOB_UPDATE)
def on_job_update(update):
    print(f'Job update: {update.job_id}, {update.status}')

@sdk.on(SDK_EVENTS.ERROR)
def on_error(error):
    print(f'Error: {error.message}')

Job Submission Examples

Examples of different job types and configurations

Simple GET Request

job_payload = JobPayload(
    request_data={
        'method': 'GET',
        'url': 'https://api.example.com/data',
        'headers': {
            'Authorization': 'Bearer your-token'
        }
    }
)

job_id = sdk.submit_job(job_payload)

Advanced Features

Advanced configuration and features

Custom Encryption

sdk = FluxumSDK(FluxumSDKOptions.builder()
    .apiKey("your-api-key")
    .enableClientSideEncryption(true)
    .encryptionKey("your-custom-encryption-key")
    .build())

Connection Management

// Manual connection
sdk.connect()

// Disconnect
sdk.disconnect()

// Check connection status
if (sdk.isConnected()) {
    println("Connected")
}

// Reconnect
sdk.reconnect()

Analytics

// Get analytics data
analytics = sdk.get_analytics()

println("Total jobs: ${analytics.total_jobs}")
println("Successful jobs: ${analytics.successful_jobs}")
println("Failed jobs: ${analytics.failed_jobs}")
println("Average response time: ${analytics.average_response_time}")

// Reset analytics
sdk.reset_analytics()

Error Handling

Comprehensive error handling and recovery

val robustEventListener = MyEventListener{}
sdk.AddEventListener(robustEventListener)

Requirements

System requirements and dependencies

Python Version

Python 3.8.0 or higher

Dependencies

  • fluxum-sdk-python (Python SDK)
  • websockets (WebSocket client)
  • cryptography (encryption)
  • asyncio (built-in Python module)

API Reference

Complete API documentation

Constructor

FluxumSDK(FluxumSDKOptions options)

class FluxumSDKOptions:
    api_key: str
    auto_connect: bool = True
    enable_client_side_encryption: bool = True
    enable_analytics: bool = True
    max_reconnect_attempts: int = 5
    reconnect_delay: int = 1000
    connection_timeout: int = 30000
    log_level: str = 'info'

Methods

// Connection
fun connect()
fun disconnect()
fun reconnect()

// Status
sdk.is_connected() -> bool
sdk.is_authenticated() -> bool

// Jobs
fun submitJob(payload: JobPayload) -> str
fun cancelJob(jobId: str) -> None

// Analytics
fun get_analytics() -> AnalyticsData
fun reset_analytics() -> None

// Events
fun addEventListener(listener: SDKEventListener)
fun removeEventListener(listener: SDKEventListener)