Python SDK Tutorial

This tutorial will guide you through using the Cloakr.ai SDK in Python applications, from basic setup to advanced features.

Installation

Prerequisites

• Python 3.8+
• pip or conda

Install the SDK

pip install cloakrai

Virtual Environment (Recommended)

# Create virtual environment
python -m venv cloakr-env

# Activate on Windows
cloakr-env\Scripts\activate

# Activate on macOS/Linux
source cloakr-env/bin/activate

# Install SDK
pip install cloakrai

Basic Setup

Initialize the Client

from cloakrai import CloakrClient
import os

client = CloakrClient(
    api_key=os.getenv('CLOAKR_API_KEY'),
    # Optional configuration
    encryption=True,
    pii_detection=True,
    retry_attempts=3,
    timeout=30
)

Environment Variables

Create a .env file in your project root:

CLOAKR_API_KEY=your-api-key-here
CLOAKR_ENCRYPTION_KEY=your-encryption-key-here  # Optional
CLOAKR_KEY_ID=your-key-id-here                  # Optional

Load environment variables:

from dotenv import load_dotenv
load_dotenv()

Basic Usage

Simple Chat Request

from cloakrai import CloakrClient
import os

client = CloakrClient(api_key=os.getenv('CLOAKR_API_KEY'))

async def simple_chat():
    try:
        response = await client.chat(
            model='gpt-4o',
            prompt='What is Cloakr.ai?',
            max_tokens=1000,
            temperature=0.7
        )
        
        print(response.choices[0].text)
    except Exception as error:
        print(f'Error: {error}')

# Run with asyncio
import asyncio
asyncio.run(simple_chat())

Synchronous Usage

from cloakrai import CloakrClient
import os

client = CloakrClient(api_key=os.getenv('CLOAKR_API_KEY'))

def simple_chat_sync():
    try:
        response = client.chat_sync(
            model='gpt-4o',
            prompt='What is Cloakr.ai?',
            max_tokens=1000,
            temperature=0.7
        )
        
        print(response.choices[0].text)
    except Exception as error:
        print(f'Error: {error}')

simple_chat_sync()

Streaming Responses

async def streaming_chat():
    try:
        async for chunk in client.chat_stream(
            model='gpt-4o',
            prompt='Write a story about AI security'
        ):
            if chunk.choices[0].delta and chunk.choices[0].delta.text:
                print(chunk.choices[0].delta.text, end='', flush=True)
    except Exception as error:
        print(f'Error: {error}')

# Run with asyncio
asyncio.run(streaming_chat())

Advanced Features

Custom PII Redaction

from cloakrai import CloakrClient, PIIRedactor
import re

client = CloakrClient(
    api_key=os.getenv('CLOAKR_API_KEY'),
    pii_detection=True
)

# Custom redaction rules
custom_redactor = PIIRedactor(
    rules=[
        {
            'type': 'custom_pattern',
            'pattern': r'CUSTOMER-\d{6}',
            'replacement': '[REDACTED_CUSTOMER_ID]'
        },
        {
            'type': 'email',
            'domains': ['internal.com', 'test.com']
        }
    ]
)

client.set_pii_redactor(custom_redactor)

Encryption Configuration

from cloakrai import CloakrClient, EncryptionConfig

encryption_config = EncryptionConfig(
    algorithm='AES-256-GCM',
    key_id=os.getenv('CLOAKR_KEY_ID'),
    key_rotation=True,
    compression=True
)

client = CloakrClient(
    api_key=os.getenv('CLOAKR_API_KEY'),
    encryption=encryption_config
)

Retry Logic with Exponential Backoff

from cloakrai import CloakrClient, RetryConfig

retry_config = RetryConfig(
    max_attempts=5,
    base_delay=1.0,
    max_delay=30.0,
    backoff_multiplier=2.0,
    retryable_errors=['rate_limit_exceeded', 'internal_error']
)

client = CloakrClient(
    api_key=os.getenv('CLOAKR_API_KEY'),
    retry=retry_config
)

Error Handling

Comprehensive Error Handling

from cloakrai import CloakrClient, CloakrError

async def robust_chat():
    try:
        response = await client.chat(
            model='gpt-4o',
            prompt='Hello world'
        )
        
        return response
    except CloakrError as error:
        if error.code == 'rate_limit_exceeded':
            print(f'Rate limited. Retry after {error.retry_after} seconds')
            # Implement retry logic
        elif error.code == 'invalid_api_key':
            print('Invalid API key. Check your configuration.')
        elif error.code == 'content_policy_violation':
            print(f'Content violates usage policies: {error.details}')
        else:
            print(f'Unexpected error: {error.message}')
        raise error
    except Exception as error:
        print(f'Network or other error: {error}')
        raise error

Custom Error Handler

import logging
from datetime import datetime

class CustomErrorHandler:
    def __init__(self):
        self.logger = logging.getLogger(__name__)
    
    async def handle(self, error, context):
        self.logger.error('Cloakr Error:', {
            'code': getattr(error, 'code', None),
            'message': str(error),
            'context': context,
            'timestamp': datetime.now().isoformat()
        })
        
        # Log to external service
        await self.log_to_external_service(error, context)
        
        # Notify team if critical
        if getattr(error, 'code', None) == 'internal_error':
            await self.notify_team(error, context)
    
    async def log_to_external_service(self, error, context):
        # Implementation for external logging
        pass
    
    async def notify_team(self, error, context):
        # Implementation for team notification
        pass

Performance Optimization

Connection Pooling

from cloakrai import CloakrClient, ConnectionPool

pool = ConnectionPool(
    max_connections=10,
    max_idle_time=30.0,
    connection_timeout=5.0
)

client = CloakrClient(
    api_key=os.getenv('CLOAKR_API_KEY'),
    connection_pool=pool
)

Request Batching

async def batch_requests(prompts):
    batch_size = 5
    results = []
    
    for i in range(0, len(prompts), batch_size):
        batch = prompts[i:i + batch_size]
        batch_tasks = [
            client.chat(model='gpt-4o', prompt=prompt)
            for prompt in batch
        ]
        
        batch_results = await asyncio.gather(*batch_tasks)
        results.extend(batch_results)
    
    return results

Caching

from cloakrai import CloakrClient, CacheManager
import hashlib

cache = CacheManager(
    ttl=3600,  # 1 hour
    max_size=1000,
    strategy='lru'
)

client = CloakrClient(
    api_key=os.getenv('CLOAKR_API_KEY'),
    cache=cache
)

# Cache key based on request parameters
def get_cache_key(model, prompt):
    content = f"{model}:{prompt}"
    return f"chat:{hashlib.md5(content.encode()).hexdigest()}"

Testing

Unit Tests with pytest

import pytest
from unittest.mock import Mock, patch
from cloakrai import CloakrClient

class TestChatService:
    @pytest.fixture
    def client(self):
        return CloakrClient(api_key='test-key')
    
    @pytest.mark.asyncio
    async def test_successful_chat_request(self, client):
        mock_response = Mock()
        mock_response.choices = [Mock(text='Mock response')]
        
        with patch.object(client, 'chat', return_value=mock_response):
            result = await client.chat(
                model='gpt-4o',
                prompt='Test prompt'
            )
            
            assert result.choices[0].text == 'Mock response'
    
    @pytest.mark.asyncio
    async def test_rate_limit_error(self, client):
        from cloakrai import CloakrError
        
        mock_error = CloakrError('Rate limit exceeded')
        mock_error.code = 'rate_limit_exceeded'
        mock_error.retry_after = 60
        
        with patch.object(client, 'chat', side_effect=mock_error):
            with pytest.raises(CloakrError):
                await client.chat(
                    model='gpt-4o',
                    prompt='Test prompt'
                )

Integration Tests

import pytest
from cloakrai import CloakrClient

class TestCloakrIntegration:
    @pytest.fixture
    def client(self):
        return CloakrClient(api_key=os.getenv('CLOAKR_TEST_API_KEY'))
    
    @pytest.mark.asyncio
    async def test_chat_request(self, client):
        response = await client.chat(
            model='gpt-4o',
            prompt='Hello, this is a test',
            max_tokens=50
        )
        
        assert hasattr(response, 'choices')
        assert hasattr(response.choices[0], 'text')
        assert response.choices[0].text

Production Best Practices

Environment Configuration

# config/cloakr.py
import os

def get_cloakr_config():
    env = os.getenv('NODE_ENV', 'development')
    
    configs = {
        'development': {
            'api_key': os.getenv('CLOAKR_DEV_API_KEY'),
            'encryption': False,
            'retry_attempts': 1
        },
        'production': {
            'api_key': os.getenv('CLOAKR_PROD_API_KEY'),
            'encryption': True,
            'retry_attempts': 3,
            'timeout': 30
        }
    }
    
    return configs.get(env, configs['development'])

Monitoring and Logging

from cloakrai import CloakrClient, MetricsCollector
import logging

class CustomMetricsCollector(MetricsCollector):
    def __init__(self, enabled=True, endpoint=None):
        super().__init__(enabled, endpoint)
        self.logger = logging.getLogger(__name__)
    
    def increment(self, metric, value=1):
        self.logger.info(f'Metric: {metric} = {value}')
    
    def timing(self, metric, duration):
        self.logger.info(f'Timing: {metric} = {duration}ms')

metrics = CustomMetricsCollector(
    enabled=os.getenv('NODE_ENV') == 'production'
)

client = CloakrClient(
    api_key=os.getenv('CLOAKR_API_KEY'),
    metrics=metrics
)

# Custom metrics
@client.on('request')
def handle_request(data):
    metrics.increment('cloakr.requests.total')
    metrics.timing('cloakr.requests.duration', data.duration)

@client.on('error')
def handle_error(error):
    metrics.increment('cloakr.errors.total')
    metrics.increment(f'cloakr.errors.{error.code}')

Security Considerations

import re

def validate_api_key(api_key):
    if not api_key or not isinstance(api_key, str):
        raise ValueError('API key must be a non-empty string')
    
    if not api_key.startswith('cloakr_'):
        raise ValueError('Invalid API key format')
    
    return api_key

# Secure client initialization
client = CloakrClient(
    api_key=validate_api_key(os.getenv('CLOAKR_API_KEY')),
    encryption=True,
    pii_detection=True
)

Async Context Manager

from contextlib import asynccontextmanager

@asynccontextmanager
async def cloakr_client():
    client = CloakrClient(api_key=os.getenv('CLOAKR_API_KEY'))
    try:
        yield client
    finally:
        await client.close()

# Usage
async def main():
    async with cloakr_client() as client:
        response = await client.chat(
            model='gpt-4o',
            prompt='Hello world'
        )
        print(response.choices[0].text)

Flask Integration

from flask import Flask, request, jsonify
from cloakrai import CloakrClient
import os

app = Flask(__name__)
client = CloakrClient(api_key=os.getenv('CLOAKR_API_KEY'))

@app.route('/chat', methods=['POST'])
async def chat():
    try:
        data = request.get_json()
        
        response = await client.chat(
            model=data.get('model', 'gpt-4o'),
            prompt=data['prompt'],
            max_tokens=data.get('max_tokens', 1000),
            temperature=data.get('temperature', 0.7)
        )
        
        return jsonify({
            'success': True,
            'response': response.choices[0].text
        })
    except Exception as error:
        return jsonify({
            'success': False,
            'error': str(error)
        }), 400

if __name__ == '__main__':
    app.run(debug=True)

FastAPI Integration

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from cloakrai import CloakrClient
import os

app = FastAPI()
client = CloakrClient(api_key=os.getenv('CLOAKR_API_KEY'))

class ChatRequest(BaseModel):
    model: str = 'gpt-4o'
    prompt: str
    max_tokens: int = 1000
    temperature: float = 0.7

class ChatResponse(BaseModel):
    success: bool
    response: str
    error: str = None

@app.post('/chat', response_model=ChatResponse)
async def chat(request: ChatRequest):
    try:
        response = await client.chat(
            model=request.model,
            prompt=request.prompt,
            max_tokens=request.max_tokens,
            temperature=request.temperature
        )
        
        return ChatResponse(
            success=True,
            response=response.choices[0].text
        )
    except Exception as error:
        raise HTTPException(status_code=400, detail=str(error))

Next Steps

• Node.js SDK Tutorial - Learn the Node.js SDK
• Java SDK Tutorial - Learn the Java SDK
• API Reference - Complete API documentation
• Security & Compliance - Enterprise security features