Add Uptime Kuma API management tooling
Implements programmatic monitor management via Socket.IO API: - manage_monitors.py: CLI script for CRUD operations on monitors - API.md: Comprehensive API usage documentation - SETUP.md: Step-by-step setup and verification guide Key features: - Supports API key and username/password authentication - List, add, update, delete monitors via CLI - SSL verification disabled for self-signed certificates - Monitor types: HTTP, TCP, ping, DNS, docker
This commit is contained in:
Executable
+156
@@ -0,0 +1,156 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Uptime Kuma Monitor Management Script
|
||||
Usage examples:
|
||||
./manage_monitors.py list
|
||||
./manage_monitors.py add --name "My Service" --url "https://example.com" --type http
|
||||
./manage_monitors.py delete --id 123
|
||||
./manage_monitors.py update --id 123 --name "Updated Name"
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
import os
|
||||
from uptime_kuma_api import UptimeKumaApi, MonitorType
|
||||
|
||||
# Configuration
|
||||
UPTIME_KUMA_URL = "https://uptime.kolpacksoftware.com"
|
||||
# Set one of these (or use environment variables):
|
||||
API_KEY = os.getenv('UPTIME_KUMA_API_KEY') # Set via UPTIME_KUMA_API_KEY env var
|
||||
USERNAME = os.getenv('UPTIME_KUMA_USERNAME') # Or use UPTIME_KUMA_USERNAME
|
||||
PASSWORD = os.getenv('UPTIME_KUMA_PASSWORD') # And UPTIME_KUMA_PASSWORD
|
||||
|
||||
def connect():
|
||||
"""Connect to Uptime Kuma API"""
|
||||
# Disable SSL verification for self-signed certificates
|
||||
api = UptimeKumaApi(UPTIME_KUMA_URL, ssl_verify=False)
|
||||
|
||||
if API_KEY:
|
||||
# API key authentication (preferred)
|
||||
api.login_by_token(API_KEY)
|
||||
elif USERNAME and PASSWORD:
|
||||
# Username/password authentication
|
||||
api.login(USERNAME, PASSWORD)
|
||||
else:
|
||||
print("Error: No credentials provided. Set UPTIME_KUMA_API_KEY or UPTIME_KUMA_USERNAME/PASSWORD environment variables.")
|
||||
sys.exit(1)
|
||||
|
||||
return api
|
||||
|
||||
def list_monitors(api):
|
||||
"""List all monitors"""
|
||||
monitors = api.get_monitors()
|
||||
if not monitors:
|
||||
print("No monitors found")
|
||||
return
|
||||
|
||||
print(f"{'ID':<6} {'Name':<30} {'Type':<10} {'URL/Hostname':<50}")
|
||||
print("-" * 96)
|
||||
for monitor in monitors:
|
||||
url = monitor.get('url') or monitor.get('hostname', 'N/A')
|
||||
if url is None:
|
||||
url = 'N/A'
|
||||
print(f"{monitor['id']:<6} {monitor['name']:<30} {str(monitor['type']):<10} {url:<50}")
|
||||
|
||||
def add_monitor(api, name, url, monitor_type='http', interval=60, ignore_tls=False):
|
||||
"""Add a new monitor"""
|
||||
type_map = {
|
||||
'http': MonitorType.HTTP,
|
||||
'https': MonitorType.HTTP,
|
||||
'tcp': MonitorType.PORT,
|
||||
'ping': MonitorType.PING,
|
||||
'dns': MonitorType.DNS,
|
||||
'docker': MonitorType.DOCKER,
|
||||
}
|
||||
|
||||
monitor_type_value = type_map.get(monitor_type.lower(), MonitorType.HTTP)
|
||||
|
||||
# Different monitor types use different parameters
|
||||
if monitor_type.lower() == 'ping':
|
||||
result = api.add_monitor(
|
||||
type=monitor_type_value,
|
||||
name=name,
|
||||
hostname=url,
|
||||
interval=interval
|
||||
)
|
||||
else:
|
||||
kwargs = {
|
||||
'type': monitor_type_value,
|
||||
'name': name,
|
||||
'url': url,
|
||||
'interval': interval
|
||||
}
|
||||
# Add SSL ignore for HTTPS URLs
|
||||
if ignore_tls or (url.startswith('https://') and monitor_type.lower() in ['http', 'https']):
|
||||
kwargs['ignoreTls'] = True
|
||||
result = api.add_monitor(**kwargs)
|
||||
print(f"Monitor added successfully: ID {result['monitorID']}")
|
||||
|
||||
def delete_monitor(api, monitor_id):
|
||||
"""Delete a monitor"""
|
||||
api.delete_monitor(monitor_id)
|
||||
print(f"Monitor {monitor_id} deleted successfully")
|
||||
|
||||
def update_monitor(api, monitor_id, **kwargs):
|
||||
"""Update a monitor"""
|
||||
# Get existing monitor
|
||||
monitor = api.get_monitor(monitor_id)
|
||||
|
||||
# Update fields
|
||||
for key, value in kwargs.items():
|
||||
if value is not None:
|
||||
monitor[key] = value
|
||||
|
||||
api.edit_monitor(monitor_id, **monitor)
|
||||
print(f"Monitor {monitor_id} updated successfully")
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description='Manage Uptime Kuma monitors')
|
||||
subparsers = parser.add_subparsers(dest='command', help='Commands')
|
||||
|
||||
# List command
|
||||
subparsers.add_parser('list', help='List all monitors')
|
||||
|
||||
# Add command
|
||||
add_parser = subparsers.add_parser('add', help='Add a monitor')
|
||||
add_parser.add_argument('--name', required=True, help='Monitor name')
|
||||
add_parser.add_argument('--url', required=True, help='URL to monitor')
|
||||
add_parser.add_argument('--type', default='http', help='Monitor type (http, tcp, ping, dns, docker)')
|
||||
add_parser.add_argument('--interval', type=int, default=60, help='Check interval in seconds')
|
||||
add_parser.add_argument('--ignore-tls', action='store_true', help='Ignore TLS/SSL certificate errors')
|
||||
|
||||
# Delete command
|
||||
delete_parser = subparsers.add_parser('delete', help='Delete a monitor')
|
||||
delete_parser.add_argument('--id', type=int, required=True, help='Monitor ID')
|
||||
|
||||
# Update command
|
||||
update_parser = subparsers.add_parser('update', help='Update a monitor')
|
||||
update_parser.add_argument('--id', type=int, required=True, help='Monitor ID')
|
||||
update_parser.add_argument('--name', help='New name')
|
||||
update_parser.add_argument('--url', help='New URL')
|
||||
update_parser.add_argument('--interval', type=int, help='New interval')
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
if not args.command:
|
||||
parser.print_help()
|
||||
sys.exit(1)
|
||||
|
||||
# Connect to API
|
||||
api = connect()
|
||||
|
||||
try:
|
||||
if args.command == 'list':
|
||||
list_monitors(api)
|
||||
elif args.command == 'add':
|
||||
ignore_tls = getattr(args, 'ignore_tls', False)
|
||||
add_monitor(api, args.name, args.url, args.type, args.interval, ignore_tls)
|
||||
elif args.command == 'delete':
|
||||
delete_monitor(api, args.id)
|
||||
elif args.command == 'update':
|
||||
update_monitor(api, args.id, name=args.name, url=args.url, interval=args.interval)
|
||||
finally:
|
||||
api.disconnect()
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user