Python Examples
Python integration examples for the CuliUptime API using session-based authentication.
📦 Installation
pip install requests
# Note: Official CuliUptime Python SDK is planned for future release
🔐 Authentication Approach
CuliUptime uses OAuth2 social login. For Python integrations, you can:
- Web Dashboard Integration: Use embedded web views for OAuth2 flow
- Session-based API: Extract session cookies from authenticated browser sessions
- Future SDK: Wait for official Python SDK with OAuth2 support
🐍 Session-Based Python Client
Basic Client Class
import requests
import json
from typing import Dict, List, Optional
from datetime import datetime
class CuliUptimeAPI:
def __init__(self, session_cookie: str, base_url: str = "https://uptime.culiops.net/api/v1"):
"""
Initialize CuliUptime API client with session cookie.
Args:
session_cookie: Session cookie value from authenticated browser
base_url: API base URL
"""
self.base_url = base_url
self.session = requests.Session()
self.session.cookies.set('session_cookie', session_cookie)
self.session.headers.update({
'Content-Type': 'application/json',
'Accept': 'application/json'
})
def _request(self, method: str, endpoint: str, **kwargs) -> Dict:
"""Make authenticated API request."""
url = f"{self.base_url}{endpoint}"
response = self.session.request(method, url, **kwargs)
response.raise_for_status()
return response.json()
# Monitor Management
def get_monitors(self, page: int = 1, per_page: int = 50, status: str = None, search: str = None) -> List[Dict]:
"""Get all monitors with optional filtering."""
params = {'page': page, 'per_page': per_page}
if status:
params['status'] = status
if search:
params['search'] = search
response = self._request('GET', '/monitors', params=params)
return response.get('data', [])
def get_monitor(self, monitor_id: int) -> Dict:
"""Get monitor details."""
response = self._request('GET', f'/monitors/{monitor_id}')
return response.get('data', {})
def create_monitor(self, monitor_data: Dict) -> Dict:
"""Create a new HTTP monitor."""
response = self._request('POST', '/monitors', json=monitor_data)
return response.get('data', {})
def update_monitor(self, monitor_id: int, monitor_data: Dict) -> Dict:
"""Update monitor configuration."""
response = self._request('PUT', f'/monitors/{monitor_id}', json=monitor_data)
return response.get('data', {})
def delete_monitor(self, monitor_id: int) -> bool:
"""Delete a monitor."""
self._request('DELETE', f'/monitors/{monitor_id}')
return True
def enable_monitor(self, monitor_id: int) -> Dict:
"""Enable a monitor."""
response = self._request('POST', f'/monitors/{monitor_id}/enable')
return response
def disable_monitor(self, monitor_id: int) -> Dict:
"""Disable a monitor."""
response = self._request('POST', f'/monitors/{monitor_id}/disable')
return response
def test_monitor(self, monitor_id: int) -> Dict:
"""Run immediate test on monitor."""
response = self._request('POST', f'/monitors/{monitor_id}/test')
return response.get('data', {})
def get_monitor_results(self, monitor_id: int, from_date: str = None, to_date: str = None) -> List[Dict]:
"""Get monitor check results."""
params = {}
if from_date:
params['from'] = from_date
if to_date:
params['to'] = to_date
response = self._request('GET', f'/monitors/{monitor_id}/results', params=params)
return response.get('data', [])
def get_monitor_stats(self, monitor_id: int) -> Dict:
"""Get monitor statistics."""
response = self._request('GET', f'/monitors/{monitor_id}/stats')
return response.get('data', {})
# Agent Management
def get_agents(self) -> List[Dict]:
"""Get all available agents."""
response = self._request('GET', '/agents')
return response.get('data', [])
def get_accessible_agents(self) -> List[Dict]:
"""Get user-accessible agents."""
response = self._request('GET', '/agents/accessible')
return response.get('data', [])
def get_agent(self, agent_id: int) -> Dict:
"""Get agent details."""
response = self._request('GET', f'/agents/{agent_id}')
return response.get('data', {})
def assign_agents_to_monitor(self, monitor_id: int, agent_ids: List[int]) -> Dict:
"""Assign agents to a monitor."""
response = self._request('POST', f'/monitors/{monitor_id}/agents',
json={'agent_ids': agent_ids})
return response
# Notifications
def get_active_notifications(self) -> List[Dict]:
"""Get currently active notifications."""
response = self._request('GET', '/notifications/active')
return response.get('data', [])
def get_recent_notifications(self) -> List[Dict]:
"""Get recent notifications."""
response = self._request('GET', '/notifications/recent')
return response.get('data', [])
def get_notifications(self, page: int = 1, limit: int = 50, monitor_id: int = None) -> List[Dict]:
"""Get notification history."""
params = {'page': page, 'limit': limit}
if monitor_id:
params['monitor_id'] = monitor_id
response = self._request('GET', '/notifications', params=params)
return response.get('data', [])
def acknowledge_notification(self, notification_id: int) -> Dict:
"""Acknowledge a notification."""
response = self._request('POST', f'/notifications/{notification_id}/acknowledge')
return response
# User & Dashboard
def get_current_user(self) -> Dict:
"""Get current user information."""
response = self._request('GET', '/auth/user')
return response.get('data', {})
def get_dashboard_data(self) -> Dict:
"""Get dashboard overview."""
response = self._request('GET', '/dashboard')
return response.get('data', {})
def get_user_preferences(self) -> Dict:
"""Get user preferences."""
response = self._request('GET', '/preferences')
return response.get('data', {})
def update_user_preferences(self, preferences: Dict) -> Dict:
"""Update user preferences."""
response = self._request('PUT', '/preferences', json=preferences)
return response.get('data', {})
🚀 Usage Examples
Initialize Client
# Initialize with session cookie from browser
api = CuliUptimeAPI(session_cookie="your_session_cookie_value")
Monitor Management Examples
# Get all monitors
monitors = api.get_monitors()
print(f"Found {len(monitors)} monitors")
# Get online monitors only
online_monitors = api.get_monitors(status="online")
# Search monitors
api_monitors = api.get_monitors(search="api")
# Create new HTTP monitor
new_monitor = api.create_monitor({
"name": "My API Endpoint",
"url": "https://api.example.com/health",
"method": "GET",
"expected_status": [200, 201],
"timeout": 30,
"check_interval": 300
})
print(f"Created monitor with ID: {new_monitor['id']}")
# Update monitor
updated_monitor = api.update_monitor(new_monitor['id'], {
"check_interval": 600,
"timeout": 45
})
# Test monitor immediately
test_result = api.test_monitor(new_monitor['id'])
print(f"Monitor status: {test_result['status']}")
# Get monitor statistics
stats = api.get_monitor_stats(new_monitor['id'])
print(f"Uptime: {stats.get('uptime_24h', 0)}%")
Notification Handling
# Check for active alerts
active_notifications = api.get_active_notifications()
if active_notifications:
print(f"⚠️ {len(active_notifications)} active notifications:")
for notification in active_notifications:
print(f" - {notification['message']}")
# Acknowledge notification
api.acknowledge_notification(notification['id'])
print(f" ✅ Acknowledged notification {notification['id']}")
# Get notification history
recent_notifications = api.get_recent_notifications()
print(f"Recent notifications: {len(recent_notifications)}")
Agent Management
# Get available agents
agents = api.get_accessible_agents()
print(f"Available agents: {len(agents)}")
for agent in agents:
print(f" - {agent['name']} ({agent['location']}) - Status: {agent['status']}")
# Assign agents to monitor
if agents and monitors:
agent_ids = [agent['id'] for agent in agents[:2]] # Use first 2 agents
monitor_id = monitors[0]['id'] # Use first monitor
result = api.assign_agents_to_monitor(monitor_id, agent_ids)
print(f"Assigned {len(agent_ids)} agents to monitor {monitor_id}")
Dashboard Integration
# Get dashboard overview
dashboard = api.get_dashboard_data()
print("📊 Dashboard Overview:")
print(f" - Total Monitors: {dashboard.get('total_monitors', 0)}")
print(f" - Online: {dashboard.get('online_monitors', 0)}")
print(f" - Offline: {dashboard.get('offline_monitors', 0)}")
print(f" - Average Uptime: {dashboard.get('average_uptime', 0):.2f}%")
🔄 Advanced Usage
Error Handling
import requests
from requests.exceptions import HTTPError, RequestException
class CuliUptimeError(Exception):
"""Custom exception for CuliUptime API errors."""
pass
# Enhanced client with error handling
def safe_api_call(func, *args, **kwargs):
"""Wrapper for safe API calls with error handling."""
try:
return func(*args, **kwargs)
except HTTPError as e:
if e.response.status_code == 401:
raise CuliUptimeError("Authentication failed - session expired")
elif e.response.status_code == 429:
raise CuliUptimeError("Rate limit exceeded")
elif e.response.status_code >= 500:
raise CuliUptimeError("Server error occurred")
else:
raise CuliUptimeError(f"API error: {e.response.status_code}")
except RequestException as e:
raise CuliUptimeError(f"Network error: {str(e)}")
# Usage with error handling
try:
monitors = safe_api_call(api.get_monitors)
print(f"Successfully retrieved {len(monitors)} monitors")
except CuliUptimeError as e:
print(f"API Error: {e}")
Batch Operations
def batch_update_monitors(api: CuliUptimeAPI, updates: List[Dict]):
"""Update multiple monitors in batch."""
results = []
for update in updates:
monitor_id = update.pop('id')
try:
result = api.update_monitor(monitor_id, update)
results.append(('success', monitor_id, result))
except Exception as e:
results.append(('error', monitor_id, str(e)))
return results
# Example batch update
updates = [
{'id': 123, 'check_interval': 300},
{'id': 124, 'timeout': 45},
{'id': 125, 'check_interval': 600}
]
results = batch_update_monitors(api, updates)
for status, monitor_id, result in results:
print(f"Monitor {monitor_id}: {status}")
Monitor Health Checker
def check_monitor_health(api: CuliUptimeAPI, monitor_id: int) -> Dict:
"""Comprehensive health check for a monitor."""
monitor = api.get_monitor(monitor_id)
stats = api.get_monitor_stats(monitor_id)
recent_results = api.get_monitor_results(monitor_id)
health_report = {
'monitor_id': monitor_id,
'name': monitor['name'],
'url': monitor['url'],
'status': monitor['status'],
'uptime_24h': stats.get('uptime_24h', 0),
'avg_response_time': stats.get('avg_response_time', 0),
'recent_failures': len([r for r in recent_results[:10] if r['status'] != 'online']),
'health_score': 'excellent' if stats.get('uptime_24h', 0) > 99 else 'good' if stats.get('uptime_24h', 0) > 95 else 'poor'
}
return health_report
# Generate health report
monitor_id = 123
health = check_monitor_health(api, monitor_id)
print(f"""
🏥 Health Report for {health['name']}:
Status: {health['status']}
24h Uptime: {health['uptime_24h']:.2f}%
Avg Response Time: {health['avg_response_time']}ms
Recent Failures: {health['recent_failures']}/10
Health Score: {health['health_score']}
""")
🔧 Session Cookie Management
Extract Session Cookie from Browser
# Manual cookie extraction (for development)
def get_session_cookie_from_browser():
"""
Instructions for extracting session cookie:
1. Open browser and login to CuliUptime dashboard
2. Open Developer Tools (F12)
3. Go to Application/Storage tab
4. Click on Cookies
5. Find 'session_cookie' and copy its value
6. Use the value to initialize the API client
"""
session_cookie = input("Enter session cookie from browser: ")
return session_cookie
# For automated browser interaction (requires selenium)
"""
from selenium import webdriver
from selenium.webdriver.common.by import By
def automated_oauth_login(email: str, password: str):
# Note: This would require implementing OAuth2 flow
# which is complex and not recommended for production
pass
"""
Session Validation
def validate_session(api: CuliUptimeAPI) -> bool:
"""Check if session is valid."""
try:
user = api.get_current_user()
return bool(user.get('id'))
except:
return False
# Check session validity
if validate_session(api):
print("✅ Session is valid")
else:
print("❌ Session expired or invalid")
Note: For production applications, consider implementing a proper OAuth2 flow or wait for the official CuliUptime Python SDK which will handle authentication securely.