Add Pydantic models and FastAPI router structure for agent and backend
CI / basic-check (push) Has been cancelled

This commit is contained in:
Perplexity Bot
2026-08-04 18:55:54 +00:00
parent dc1b2ab99b
commit 69124b31f1
15 changed files with 546 additions and 111 deletions
+21
View File
@@ -0,0 +1,21 @@
from fastapi import APIRouter
from app.models import HealthResponse
from app.services.runtime_probe import build_health
router = APIRouter(tags=['health'])
@router.get('/live', response_model=HealthResponse)
def live() -> HealthResponse:
return build_health()
@router.get('/ready', response_model=HealthResponse)
def ready() -> HealthResponse:
return build_health()
@router.get('/health', response_model=HealthResponse)
def health() -> HealthResponse:
return build_health()
+13
View File
@@ -0,0 +1,13 @@
from fastapi import APIRouter
from app.models import RuntimeResponse
from app.services.runtime_probe import build_runtime
from app.services.systemd_manager import SystemdManager
router = APIRouter(tags=['runtime'])
manager = SystemdManager()
@router.get('/runtime', response_model=RuntimeResponse)
def runtime() -> RuntimeResponse:
return build_runtime(manager)
+42
View File
@@ -0,0 +1,42 @@
from datetime import datetime, timezone
from typing import Literal
from fastapi import APIRouter, HTTPException
from app.config import settings
from app.models import ServiceActionResponse, ServiceListResponse, ServiceState
from app.services.systemd_manager import SystemdManager
router = APIRouter(prefix='/services', tags=['services'])
manager = SystemdManager()
@router.get('', response_model=ServiceListResponse)
def list_services() -> ServiceListResponse:
items = manager.list_services(settings.monitored_services)
return ServiceListResponse(count=len(items), time=datetime.now(timezone.utc), items=items)
@router.get('/{name}', response_model=ServiceState)
def get_service(name: str) -> ServiceState:
items = manager.list_services([name])
if not items:
raise HTTPException(status_code=404, detail='Service not found')
return items[0]
@router.post('/{name}/{action}', response_model=ServiceActionResponse)
def service_action(name: str, action: Literal['start', 'stop', 'restart']) -> ServiceActionResponse:
if action == 'start':
manager.start(name)
elif action == 'stop':
manager.stop(name)
elif action == 'restart':
manager.restart(name)
return ServiceActionResponse(
status='accepted',
action=action,
service=name,
time=datetime.now(timezone.utc),
message=f'{action} requested',
)