Python SDK

The Connix Python SDK provides a Pythonic interface for integrating Connix services into your Python applications. Built for data science, machine learning, and automation workflows with excellent async support and type safety.

Requirements

Python 3.7 or later
pip or poetry for package management

Installation

Using pip

pip install connix-sdk

Using poetry

poetry add connix-sdk

Using conda

conda install -c connix connix-sdk

Quick Start

import connix

# Initialize the client
client = connix.Client("your-api-key")

# Create a project for your ML workflow
try:
    project = client.create_project(name="sentiment-analysis")
    print(f"✅ Created project: {project.name} (ID: {project.id})")

    # Launch an agent for data processing
    agent = client.launch_agent(
        name="data-processor",
        project_id=project.id
    )
    print(f"🚀 Agent launched: {agent.message}")

    # List all projects
    projects = client.list_projects()
    print(f"📋 Total projects: {len(projects.projects)}")

except connix.ConnixError as e:
    print(f"Error: {e}")

Configuration

Configure the SDK with various options:

import connix
from datetime import timedelta

# Using configuration object
config = connix.Config(
    timeout=timedelta(seconds=30),
    max_retries=3,
    base_url="https://api.connix.io",
    headers={"Custom-Header": "value"}
)

client = connix.Client("your-api-key", config=config)

# Or using keyword arguments
client = connix.Client(
    "your-api-key",
    timeout=30,
    max_retries=3,
    base_url="https://api.connix.io"
)

Async Support

The SDK supports both synchronous and asynchronous operations:

import asyncio
import connix

# Async client
async def main():
    async with connix.AsyncClient("your-api-key") as client:
        try:
            result = await client.some_method("parameter")
            print(f"Async result: {result.data}")
        except connix.ConnixError as e:
            print(f"Error: {e}")

# Run async function
asyncio.run(main())

Error Handling

The SDK provides structured exception handling:

import connix

client = connix.Client("your-api-key")

try:
    result = client.some_method("parameter")
except connix.ValidationError as e:
    print(f"Validation error: {e}")
    print(f"Details: {e.validation_errors}")
except connix.NetworkError as e:
    print(f"Network error: {e}")
except connix.ConnixError as e:
    print(f"Connix error: {e} (code: {e.error_code})")
except Exception as e:
    print(f"Unexpected error: {e}")

Context Manager Support

Use the client as a context manager for automatic resource cleanup:

import connix

with connix.Client("your-api-key") as client:
    result = client.get_data()
    print(f"Data: {result}")
# Client is automatically closed here

Type Hints

The SDK includes comprehensive type hints for better IDE support:

from typing import List
import connix

client: connix.Client = connix.Client("your-api-key")

def process_data(client: connix.Client) -> List[connix.DataItem]:
    response: connix.DataResponse = client.get_data()
    return response.items

API Reference

Client

Constructor

def __init__(
    self,
    api_key: str,
    config: Optional[Config] = None,
    timeout: Optional[int] = None,
    max_retries: Optional[int] = None,
    base_url: Optional[str] = None
) -> None

Methods

def some_method(self, param: str) -> Result
def get_data(self) -> DataResponse
def close(self) -> None

AsyncClient

Constructor

def __init__(
    self,
    api_key: str,
    config: Optional[Config] = None,
    **kwargs
) -> None

Methods

async def some_method(self, param: str) -> Result
async def get_data(self) -> DataResponse
async def close(self) -> None

Configuration

Config

@dataclass
class Config:
    timeout: timedelta = timedelta(seconds=30)
    max_retries: int = 3
    base_url: str = "https://api.connix.io"
    headers: Dict[str, str] = field(default_factory=dict)

Result

@dataclass
class Result:
    id: str
    data: str
    timestamp: datetime

Django Integration

For Django projects, add configuration to your settings:

CONNIX = {
    'API_KEY': 'your-api-key',
    'TIMEOUT': 30,
    'MAX_RETRIES': 3,
    'BASE_URL': 'https://api.connix.io',
}

from django.conf import settings
import connix

def my_view(request):
    client = connix.Client(settings.CONNIX['API_KEY'])
    result = client.get_data()
    return JsonResponse({'data': result.data})

Flask Integration

from flask import Flask, jsonify
import connix

app = Flask(__name__)
app.config['CONNIX_API_KEY'] = 'your-api-key'

# Initialize client
connix_client = connix.Client(app.config['CONNIX_API_KEY'])

@app.route('/data')
def get_data():
    try:
        result = connix_client.get_data()
        return jsonify({'data': result.data})
    except connix.ConnixError as e:
        return jsonify({'error': str(e)}), 500

Examples

Basic Usage

import connix

def main():
    client = connix.Client("your-api-key")

    try:
        response = client.get_data()
        for item in response.items:
            print(f"Item: {item.name}")
    except connix.ConnixError as e:
        print(f"Error: {e}")
    finally:
        client.close()

if __name__ == "__main__":
    main()

Async Batch Processing

import asyncio
import connix

async def process_batch(items):
    async with connix.AsyncClient("your-api-key") as client:
        tasks = [client.process_item(item) for item in items]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        for i, result in enumerate(results):
            if isinstance(result, Exception):
                print(f"Item {i} failed: {result}")
            else:
                print(f"Item {i} processed: {result.status}")

# Usage
items = ["item1", "item2", "item3"]
asyncio.run(process_batch(items))

With Custom Configuration

import connix
from datetime import timedelta

def main():
    config = connix.Config(
        timeout=timedelta(minutes=1),
        max_retries=5,
        headers={"User-Agent": "MyApp/1.0"}
    )

    with connix.Client("your-api-key", config=config) as client:
        try:
            result = client.process_large_data("large-dataset")
            print(f"Processing complete: {result.status}")
        except connix.ConnixError as e:
            print(f"Error: {e}")

if __name__ == "__main__":
    main()

Logging Integration

import logging
import connix

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Enable SDK logging
connix.enable_logging(level=logging.DEBUG)

def main():
    client = connix.Client("your-api-key")

    try:
        result = client.get_data()
        logger.info(f"Retrieved {len(result.items)} items")
    except connix.ConnixError as e:
        logger.error(f"Failed to get data: {e}")

if __name__ == "__main__":
    main()

Support

For support and questions, visit our GitHub repository or contact support.