Python SDK
The Unizo Python SDK provides a convenient way to interact with the Unizo API from Python applications. It supports Python 3.7+ and includes type hints for better development experience.
Installation
pip
pip install unizo-sdk
Poetry
poetry add unizo-sdk
pipenv
pipenv install unizo-sdk
Requirements File
# requirements.txt
unizo-sdk>=1.0.0
Basic Setup
from unizo import UnizoClient
# Initialize the client
client = UnizoClient(
api_key="your-api-key",
environment="production" # or "sandbox"
)
Environment Variables
import os
from unizo import UnizoClient
# Using environment variables
client = UnizoClient(
api_key=os.getenv("UNIZO_API_KEY"),
environment=os.getenv("UNIZO_ENVIRONMENT", "production")
)
Configuration Options
from unizo import UnizoClient
from unizo.config import RetryConfig
client = UnizoClient(
api_key="your-api-key",
environment="production",
timeout=30.0, # Request timeout in seconds
base_url="https://api.unizo.com", # Custom base URL
user_agent="MyApp/1.0.0", # Custom user agent
retry_config=RetryConfig(
attempts=3,
delay=1.0,
backoff_factor=2.0
),
debug=False # Enable debug logging
)
Authentication
API Key Authentication
# Set during initialization
client = UnizoClient(api_key="your-api-key")
# Or set later
client.set_api_key("your-api-key")
# Using context manager for temporary auth
with client.with_api_key("temp-api-key") as temp_client:
result = temp_client.identity.get()
Bearer Token Authentication
# For user-specific operations
client = UnizoClient(bearer_token="user-access-token")
# Or set dynamically
client.set_bearer_token("user-access-token")
Usage Examples
Identity Management
from unizo.exceptions import UnizoError
try:
# Get user identity
identity = client.identity.get()
print(f"User ID: {identity.id}")
print(f"Email: {identity.email}")
# Update user profile
updated_profile = client.identity.update(
name="John Doe",
email="john@example.com",
metadata={
"department": "Engineering",
"role": "Senior Developer"
}
)
except UnizoError as e:
print(f"Error: {e.message}")
Source Code Management
from unizo.types import WebhookEvent
# List repositories
repositories = client.source_code.repositories.list(
page=1,
limit=50,
provider="github"
)
for repo in repositories.data:
print(f"Repository: {repo.name} ({repo.url})")
# Get repository details
repo = client.source_code.repositories.get("repo-id")
print(f"Repository stats: {repo.stats}")
# Create webhook
webhook = client.source_code.webhooks.create(
repository_id="repo-id",
url="https://your-app.com/webhooks/source-code",
events=[WebhookEvent.PUSH, WebhookEvent.PULL_REQUEST],
secret="webhook-secret"
)
Ticketing System
from unizo.types import TicketPriority, TicketStatus
from datetime import datetime, timedelta
# Create a ticket
ticket = client.ticketing.tickets.create(
title="Bug Report: Login Issues",
description="Users are experiencing login failures after the latest update",
priority=TicketPriority.HIGH,
assignee="user-id",
tags=["bug", "login", "urgent"],
due_date=datetime.now() + timedelta(days=3)
)
print(f"Created ticket: {ticket.id}")
# List tickets with filtering
tickets = client.ticketing.tickets.list(
status=TicketStatus.OPEN,
priority=TicketPriority.HIGH,
assignee="current-user",
page=1,
limit=25,
created_after=datetime.now() - timedelta(days=7)
)
print(f"Found {tickets.total} tickets")
# Update ticket
updated_ticket = client.ticketing.tickets.update(
ticket_id="ticket-id",
status=TicketStatus.IN_PROGRESS,
comments=[{
"text": "Started investigating the issue",
"author": "developer-id",
"timestamp": datetime.now()
}]
)
Communications
from unizo.types import EmailTemplate
# Send email with template
email_result = client.communications.email.send(
to=["user@example.com"],
subject="Welcome to Unizo",
template="welcome-template",
variables={
"user_name": "John Doe",
"activation_link": "https://app.unizo.com/activate/token"
},
attachments=[
{
"filename": "welcome_guide.pdf",
"content": b"PDF content here",
"content_type": "application/pdf"
}
]
)
# Send plain text email
plain_email = client.communications.email.send(
to=["user@example.com"],
subject="System Notification",
text="Your account has been activated successfully."
)
# Send SMS
sms_result = client.communications.sms.send(
to="+1234567890",
message="Your verification code is: 123456",
sender_id="UNIZO"
)
print(f"SMS sent: {sms_result.message_id}")
Incident Management
from unizo.types import IncidentSeverity, IncidentStatus
# Create incident
incident = client.incidents.create(
title="API Service Degradation",
description="Response times increased by 200% across all endpoints",
severity=IncidentSeverity.MAJOR,
status=IncidentStatus.INVESTIGATING,
services=["api-gateway", "database", "cache"],
affected_users=1500
)
# List active incidents
active_incidents = client.incidents.list(
status=[IncidentStatus.OPEN, IncidentStatus.INVESTIGATING],
severity=[IncidentSeverity.CRITICAL, IncidentSeverity.MAJOR]
)
# Update incident with timeline
client.incidents.update(
incident_id="incident-id",
status=IncidentStatus.RESOLVED,
resolution="Scaled up database instances and optimized queries",
timeline=[
{
"timestamp": datetime.now(),
"event": "Issue resolved",
"description": "Database performance restored to normal levels"
}
]
)
Error Handling
Exception Types
from unizo.exceptions import (
UnizoError,
ValidationError,
AuthenticationError,
RateLimitError,
NotFoundError,
ServerError
)
try:
result = client.identity.get()
except ValidationError as e:
print(f"Validation error: {e.message}")
print(f"Field errors: {e.field_errors}")
except AuthenticationError as e:
print(f"Authentication failed: {e.message}")
# Handle re-authentication
except RateLimitError as e:
print(f"Rate limit exceeded. Retry after: {e.retry_after} seconds")
# Implement backoff strategy
except NotFoundError as e:
print(f"Resource not found: {e.message}")
except ServerError as e:
print(f"Server error: {e.message}")
# Log for debugging
except UnizoError as e:
print(f"Unizo API error: {e.message}")
print(f"Status code: {e.status_code}")
print(f"Error code: {e.code}")
Retry Decorators
from functools import wraps
import time
import random
def retry_on_rate_limit(max_retries=3, base_delay=1.0):
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except RateLimitError as e:
if attempt == max_retries - 1:
raise
# Exponential backoff with jitter
delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
time.sleep(min(delay, e.retry_after))
return func(*args, **kwargs)
return wrapper
return decorator
@retry_on_rate_limit(max_retries=3)
def get_tickets():
return client.ticketing.tickets.list()
Async Support
import asyncio
from unizo import AsyncUnizoClient
async def main():
async_client = AsyncUnizoClient(api_key="your-api-key")
try:
# Async operations
identity = await async_client.identity.get()
# Concurrent requests
tasks = [
async_client.ticketing.tickets.get(ticket_id)
for ticket_id in ["ticket-1", "ticket-2", "ticket-3"]
]
tickets = await asyncio.gather(*tasks)
finally:
await async_client.close()
# Run async code
asyncio.run(main())
Async Context Manager
async def main():
async with AsyncUnizoClient(api_key="your-api-key") as client:
identity = await client.identity.get()
tickets = await client.ticketing.tickets.list()
Webhook Handling
Flask Example
from flask import Flask, request, jsonify
from unizo.webhooks import validate_webhook_signature
import os
app = Flask(__name__)
@app.route('/webhooks/unizo', methods=['POST'])
def handle_webhook():
signature = request.headers.get('X-Unizo-Signature')
payload = request.get_data()
if not validate_webhook_signature(
payload,
signature,
os.getenv('WEBHOOK_SECRET')
):
return jsonify({'error': 'Invalid signature'}), 401
event = request.get_json()
if event['type'] == 'ticket.created':
handle_ticket_created(event['data'])
elif event['type'] == 'incident.updated':
handle_incident_updated(event['data'])
return jsonify({'status': 'success'})
def handle_ticket_created(ticket_data):
print(f"New ticket created: {ticket_data['id']}")
# Process ticket creation
def handle_incident_updated(incident_data):
print(f"Incident updated: {incident_data['id']}")
# Process incident update
FastAPI Example
from fastapi import FastAPI, Request, HTTPException, Header
from unizo.webhooks import validate_webhook_signature
import os
app = FastAPI()
@app.post("/webhooks/unizo")
async def handle_webhook(
request: Request,
x_unizo_signature: str = Header(None)
):
payload = await request.body()
if not validate_webhook_signature(
payload,
x_unizo_signature,
os.getenv('WEBHOOK_SECRET')
):
raise HTTPException(status_code=401, detail="Invalid signature")
event = await request.json()
# Process webhook event
await process_webhook_event(event)
return {"status": "success"}
Type Hints and Data Classes
from typing import List, Optional, Dict, Any
from dataclasses import dataclass
from unizo.types import Ticket, User, CreateTicketRequest
@dataclass
class TicketAnalytics:
total_tickets: int
open_tickets: int
resolved_tickets: int
average_resolution_time: float
def analyze_tickets(tickets: List[Ticket]) -> TicketAnalytics:
total = len(tickets)
open_count = sum(1 for t in tickets if t.status == "open")
resolved_count = sum(1 for t in tickets if t.status == "resolved")
return TicketAnalytics(
total_tickets=total,
open_tickets=open_count,
resolved_tickets=resolved_count,
average_resolution_time=calculate_avg_time(tickets)
)
# Type-safe ticket creation
def create_bug_ticket(
title: str,
description: str,
assignee: Optional[str] = None
) -> Ticket:
request = CreateTicketRequest(
title=title,
description=description,
priority="high",
tags=["bug"],
assignee=assignee
)
return client.ticketing.tickets.create(request)
Testing
Using pytest
import pytest
from unittest.mock import Mock, patch
from unizo import UnizoClient
from unizo.exceptions import ValidationError
@pytest.fixture
def client():
return UnizoClient(api_key="test-api-key")
@pytest.fixture
def mock_response():
return {
"id": "ticket-123",
"title": "Test Ticket",
"status": "open"
}
def test_create_ticket_success(client, mock_response):
with patch.object(client.ticketing.tickets, 'create') as mock_create:
mock_create.return_value = mock_response
result = client.ticketing.tickets.create(
title="Test Ticket",
description="Test Description"
)
assert result["id"] == "ticket-123"
mock_create.assert_called_once()
def test_create_ticket_validation_error(client):
with patch.object(client.ticketing.tickets, 'create') as mock_create:
mock_create.side_effect = ValidationError("Title is required")
with pytest.raises(ValidationError):
client.ticketing.tickets.create(title="", description="Test")
Mock Client for Testing
from unizo.testing import MockUnizoClient
def test_with_mock_client():
mock_client = MockUnizoClient()
# Set up mock responses
mock_client.ticketing.tickets.create.return_value = {
"id": "mock-ticket-id",
"title": "Mock Ticket"
}
# Use in your code
result = mock_client.ticketing.tickets.create(
title="Test Ticket",
description="Test Description"
)
assert result["id"] == "mock-ticket-id"
Logging and Debugging
import logging
from unizo import UnizoClient
# Enable debug logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger('unizo')
client = UnizoClient(
api_key="your-api-key",
debug=True # Enable request/response logging
)
# Custom logging
def log_api_calls(func):
def wrapper(*args, **kwargs):
logger.info(f"Calling {func.__name__} with args: {args}, kwargs: {kwargs}")
try:
result = func(*args, **kwargs)
logger.info(f"Success: {func.__name__}")
return result
except Exception as e:
logger.error(f"Error in {func.__name__}: {e}")
raise
return wrapper
# Apply to client methods
client.identity.get = log_api_calls(client.identity.get)
Performance Optimization
Connection Pooling
from unizo import UnizoClient
import requests.adapters
# Configure connection pooling
session = requests.Session()
adapter = requests.adapters.HTTPAdapter(
pool_connections=20,
pool_maxsize=20,
max_retries=3
)
session.mount('https://', adapter)
client = UnizoClient(
api_key="your-api-key",
session=session
)
Batch Operations
# Batch ticket creation
tickets_data = [
{"title": "Issue 1", "description": "First issue"},
{"title": "Issue 2", "description": "Second issue"},
{"title": "Issue 3", "description": "Third issue"}
]
# Process in batches
batch_size = 10
for i in range(0, len(tickets_data), batch_size):
batch = tickets_data[i:i + batch_size]
results = client.ticketing.tickets.create_batch(batch)
print(f"Created {len(results)} tickets")
Django Integration
# settings.py
UNIZO_API_KEY = os.getenv('UNIZO_API_KEY')
UNIZO_ENVIRONMENT = os.getenv('UNIZO_ENVIRONMENT', 'production')
# Create a Django service
from django.conf import settings
from unizo import UnizoClient
class UnizoService:
def __init__(self):
self.client = UnizoClient(
api_key=settings.UNIZO_API_KEY,
environment=settings.UNIZO_ENVIRONMENT
)
def create_support_ticket(self, user, subject, message):
return self.client.ticketing.tickets.create(
title=subject,
description=message,
assignee=None,
metadata={
'user_id': user.id,
'user_email': user.email
}
)
# Use in views
from django.shortcuts import render
from .services import UnizoService
def submit_support_ticket(request):
if request.method == 'POST':
unizo_service = UnizoService()
ticket = unizo_service.create_support_ticket(
user=request.user,
subject=request.POST['subject'],
message=request.POST['message']
)
return render(request, 'success.html', {'ticket_id': ticket.id})
Resources
Support
- Create issues on GitHub
- Email: sdk-support@unizo.com
- Discord: Unizo Developers
Requirements
- Python 3.7+
- requests >= 2.25.0
- typing-extensions >= 3.7.4 (for Python < 3.8)
- pydantic >= 1.8.0 (for data validation)