""" Base de données hybride combinant données Ansible et données runtime. """ from datetime import datetime, timezone from typing import Any, Dict, List, Optional from app.schemas.host_api import Host from app.schemas.task_api import Task from app.schemas.common import LogEntry, SystemMetrics class HybridDB: """Base de données hybride combinant l'inventaire Ansible avec les données runtime. Cette classe agit comme un agrégateur de données provenant de plusieurs sources: - Inventaire Ansible (via AnsibleService) - Statuts bootstrap (via BootstrapStatusService) - Statuts runtime (via HostStatusService) - Tâches et logs en mémoire """ def __init__(self): # Cache des hôtes self._hosts_cache: Optional[List[Host]] = None self._hosts_cache_time: float = 0 self._cache_ttl = 60 # secondes # Données en mémoire self.tasks: List[Task] = [] self.logs: List[LogEntry] = [] # Compteurs pour les IDs self._id_counters: Dict[str, int] = { "tasks": 0, "logs": 0, } def get_next_id(self, entity: str) -> int: """Génère un nouvel ID pour une entité.""" self._id_counters[entity] = self._id_counters.get(entity, 0) + 1 return self._id_counters[entity] @property def hosts(self) -> List[Host]: """Retourne la liste des hôtes, en la mettant à jour si nécessaire.""" import time current_time = time.time() if self._hosts_cache and (current_time - self._hosts_cache_time) < self._cache_ttl: return self._hosts_cache return self.refresh_hosts() def refresh_hosts(self) -> List[Host]: """Rafraîchit la liste des hôtes depuis l'inventaire Ansible.""" import time from app.services.ansible_service import ansible_service from app.services.bootstrap_status_service import bootstrap_status_service from app.services.host_status_service import host_status_service hosts = [] inventory_hosts = ansible_service.get_hosts_from_inventory() for inv_host in inventory_hosts: # Récupérer le statut bootstrap bs_status = bootstrap_status_service.get_bootstrap_status(inv_host.name) # Récupérer le statut runtime rt_status = host_status_service.get_status(inv_host.name) host = Host( id=inv_host.name, # Utiliser le nom comme ID name=inv_host.name, ip=inv_host.ansible_host or inv_host.name, status=rt_status.get("status") or "unknown", os=rt_status.get("os") or "Linux", last_seen=rt_status.get("last_seen"), groups=inv_host.groups or [inv_host.group] if inv_host.group else [], bootstrap_ok=bs_status.get("bootstrap_ok", False), bootstrap_date=bs_status.get("bootstrap_date") ) hosts.append(host) self._hosts_cache = hosts self._hosts_cache_time = time.time() return hosts def invalidate_hosts_cache(self): """Invalide le cache des hôtes.""" self._hosts_cache = None def get_host(self, host_id: str) -> Optional[Host]: """Récupère un hôte par son ID ou nom.""" for host in self.hosts: if host.id == host_id or host.name == host_id or host.ip == host_id: return host return None def update_host_status( self, host_name: str, status: str, os_info: str = None ): """Met à jour le statut d'un hôte.""" from app.services.host_status_service import host_status_service host_status_service.set_status( host_name=host_name, status=status, last_seen=datetime.now(timezone.utc), os_info=os_info ) # Invalider le cache pour forcer le rechargement self.invalidate_hosts_cache() @property def metrics(self) -> SystemMetrics: """Calcule et retourne les métriques système.""" hosts = self.hosts online_count = sum(1 for h in hosts if h.status == "online") total_tasks = len(self.tasks) # Calculer le taux de succès completed = sum(1 for t in self.tasks if t.status == "completed") failed = sum(1 for t in self.tasks if t.status == "failed") total_finished = completed + failed success_rate = (completed / total_finished * 100) if total_finished > 0 else 100.0 return SystemMetrics( online_hosts=online_count, total_tasks=total_tasks, success_rate=round(success_rate, 1), uptime=99.9, # TODO: calculer depuis le démarrage cpu_usage=0.0, memory_usage=0.0, disk_usage=0.0 ) def add_task(self, task: Task): """Ajoute une tâche à la liste.""" self.tasks.insert(0, task) # Limiter la taille de la liste if len(self.tasks) > 1000: self.tasks = self.tasks[:1000] def get_task(self, task_id: str) -> Optional[Task]: """Récupère une tâche par son ID.""" for task in self.tasks: if str(task.id) == str(task_id): return task return None def update_task(self, task_id: str, **kwargs): """Met à jour une tâche existante.""" task = self.get_task(task_id) if task: for key, value in kwargs.items(): if hasattr(task, key): setattr(task, key, value) def add_log(self, log: LogEntry): """Ajoute une entrée de log.""" if log.id == 0: log.id = self.get_next_id("logs") self.logs.insert(0, log) # Limiter la taille de la liste if len(self.logs) > 5000: self.logs = self.logs[:5000] def get_recent_logs(self, limit: int = 50, level: str = None, source: str = None) -> List[LogEntry]: """Récupère les logs récents avec filtrage optionnel.""" logs = self.logs if level: logs = [l for l in logs if l.level == level] if source: logs = [l for l in logs if l.source == source] return logs[:limit] def clear_logs(self): """Efface tous les logs.""" self.logs.clear() def clear_tasks(self): """Efface toutes les tâches.""" self.tasks.clear() # Instance singleton de la base de données hybride db = HybridDB()