Files
core-api/src/service_groups.py
T
jpmschweitzerandClaude Opus 4.5 ac92607003 Remove Uptime Kuma integration
- Delete kuma_client.py and all Kuma-related code
- Remove /infrastructure/monitors endpoints
- Update service start/stop to use Portainer only
- Simplify service control widget to show container status
- Remove Kuma config and credentials
- Delete stale memory tests (moved to core-ai service)
- Add test infrastructure with pytest
- Add tests for service_groups, config, and health endpoints
- Bump version to 1.1.0

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-14 12:04:05 +01:00

136 lines
3.1 KiB
Python

"""
Service Groups and Safety Configuration
Defines service groups, dependencies, and always-on infrastructure services.
"""
from typing import List, Dict, Set
# Always-on infrastructure services (CANNOT be stopped via API)
ALWAYS_ON_SERVICES: Set[str] = {
"portainer",
"nginx-proxy-manager",
"core-api",
"organizr",
"headscale",
"watchtower",
"netdata",
"maintenance",
"postgres-shared",
"redis-shared",
"authentik",
}
# Service groups - services that should be started/stopped together
SERVICE_GROUPS: Dict[str, List[str]] = {
"jellyfin": [
"jellyfin",
],
"nextcloud": [
"nextcloud",
],
"gitea": [
"gitea",
"gitea-db",
],
"ai-stack": [
"open-webui",
"ollama",
"qdrant",
],
"samba": [
"samba",
],
}
# Reverse mapping: service name -> group name
SERVICE_TO_GROUP: Dict[str, str] = {}
for group, services in SERVICE_GROUPS.items():
for service in services:
SERVICE_TO_GROUP[service] = group
def is_always_on(service_name: str) -> bool:
"""
Check if a service is marked as always-on (infrastructure)
Args:
service_name: Name of the service
Returns:
True if service cannot be stopped, False otherwise
"""
return service_name.lower() in ALWAYS_ON_SERVICES
def get_service_group(service_name: str) -> List[str]:
"""
Get all services in the same group as the given service
Args:
service_name: Name of the service
Returns:
List of service names in the group (including the service itself)
Returns [service_name] if not part of a group
"""
group = SERVICE_TO_GROUP.get(service_name.lower())
if group:
return SERVICE_GROUPS[group].copy()
return [service_name]
def get_group_name(service_name: str) -> str:
"""
Get the group name for a service
Args:
service_name: Name of the service
Returns:
Group name or the service name if not in a group
"""
return SERVICE_TO_GROUP.get(service_name.lower(), service_name)
def list_service_groups() -> Dict[str, List[str]]:
"""
Get all defined service groups
Returns:
Dictionary of group names to service lists
"""
return SERVICE_GROUPS.copy()
def list_stoppable_services() -> List[str]:
"""
Get list of all services that can be stopped
Returns:
List of service names that are not always-on
"""
stoppable = []
for services in SERVICE_GROUPS.values():
stoppable.extend(services)
# Remove any always-on services (shouldn't be in groups, but safety check)
return [s for s in stoppable if not is_always_on(s)]
def validate_stop_request(service_names: List[str]) -> tuple[bool, str]:
"""
Validate that a list of services can be stopped
Args:
service_names: List of service names to check
Returns:
Tuple of (is_valid, error_message)
error_message is empty string if valid
"""
for service in service_names:
if is_always_on(service):
return False, f"Cannot stop always-on service: {service}"
return True, ""