Module demo_api.utils.services

Services métier pour demo_api

Ce module contient les services qui encapsulent la logique métier : - VMService : Gestion des VMs - ReportService : Génération de rapports - DataManager : Gestion centralisée des données

Sub-modules

demo_api.utils.services.data_manager

Gestionnaire de données centralisé pour éviter les requêtes multiples

demo_api.utils.services.report_service

Service de génération de rapports

demo_api.utils.services.vm_service

Service de gestion des VMs

Classes

class DataManager (api_client: utils.api.ApiClient)

Expand source code

class DataManager:
    """Gestionnaire centralisé des données pour éviter les requêtes multiples"""

    def __init__(self, api_client: Api):
        """
        Initialise le gestionnaire de données

        Args:
            api_client: Client API unifié
        """
        self.api = api_client
        self._users_cache: Optional[List[Dict[str, Any]]] = None
        self._vms_cache: Optional[List[Dict[str, Any]]] = None
        self._data_fetched = False

    def fetch_all_data(self) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
        """
        Récupère toutes les données nécessaires (utilisateurs et VMs) en une seule fois

        Returns:
            Tuple contenant (users, vms) ou ([], []) en cas d'erreur
        """
        if self._data_fetched:
            logger.info("Utilisation des données en cache")
            return self._users_cache or [], self._vms_cache or []

        logger.info("Début de récupération centralisée des données")

        # Récupération des utilisateurs
        try:
            self._users_cache = self.api.users.get()
            logger.info("Utilisateurs récupérés", count=len(self._users_cache))
        except UsersFetchError as e:
            logger.error("Impossible de récupérer les utilisateurs", error=str(e))
            self._users_cache = []

        # Récupération des VMs
        try:
            self._vms_cache = self.api.vms.get()
            logger.info("VMs récupérées", count=len(self._vms_cache))
        except VMsFetchError as e:
            logger.error("Impossible de récupérer les VMs", error=str(e))
            self._vms_cache = []

        self._data_fetched = True
        logger.info("Récupération centralisée des données terminée")

        return self._users_cache or [], self._vms_cache or []

    def get_users(self) -> List[Dict[str, Any]]:
        """
        Retourne les utilisateurs (doit être appelé après fetch_all_data)

        Returns:
            Liste des utilisateurs
        """
        if not self._data_fetched:
            raise RuntimeError("fetch_all_data() doit être appelé avant get_users()")
        return self._users_cache or []

    def get_vms(self) -> List[Dict[str, Any]]:
        """
        Retourne les VMs (doit être appelé après fetch_all_data)

        Returns:
            Liste des VMs
        """
        if not self._data_fetched:
            raise RuntimeError("fetch_all_data() doit être appelé avant get_vms()")
        return self._vms_cache or []

    def get_users_with_vms(self) -> List[Dict[str, Any]]:
        """
        Retourne les utilisateurs avec leurs VMs associées

        Returns:
            Liste des utilisateurs avec leurs VMs
        """
        users = self.get_users()
        vms = self.get_vms()

        # Associer les VMs aux utilisateurs
        self.api.users.add_vms_to_users(users, vms)
        return users

    def clear_cache(self) -> None:
        """Vide le cache des données"""
        self._users_cache = None
        self._vms_cache = None
        self._data_fetched = False
        logger.info("Cache des données vidé")

    @property
    def is_data_loaded(self) -> bool:
        """Indique si les données sont chargées"""
        return self._data_fetched

Gestionnaire centralisé des données pour éviter les requêtes multiples

Initialise le gestionnaire de données

Args

api_client

Client API unifié

Instance variables

prop is_data_loaded : bool

Expand source code

@property
def is_data_loaded(self) -> bool:
    """Indique si les données sont chargées"""
    return self._data_fetched

Indique si les données sont chargées

Methods

def clear_cache(self) ‑> None

Expand source code

def clear_cache(self) -> None:
    """Vide le cache des données"""
    self._users_cache = None
    self._vms_cache = None
    self._data_fetched = False
    logger.info("Cache des données vidé")

Vide le cache des données

def fetch_all_data(self) ‑> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]

Expand source code

def fetch_all_data(self) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]:
    """
    Récupère toutes les données nécessaires (utilisateurs et VMs) en une seule fois

    Returns:
        Tuple contenant (users, vms) ou ([], []) en cas d'erreur
    """
    if self._data_fetched:
        logger.info("Utilisation des données en cache")
        return self._users_cache or [], self._vms_cache or []

    logger.info("Début de récupération centralisée des données")

    # Récupération des utilisateurs
    try:
        self._users_cache = self.api.users.get()
        logger.info("Utilisateurs récupérés", count=len(self._users_cache))
    except UsersFetchError as e:
        logger.error("Impossible de récupérer les utilisateurs", error=str(e))
        self._users_cache = []

    # Récupération des VMs
    try:
        self._vms_cache = self.api.vms.get()
        logger.info("VMs récupérées", count=len(self._vms_cache))
    except VMsFetchError as e:
        logger.error("Impossible de récupérer les VMs", error=str(e))
        self._vms_cache = []

    self._data_fetched = True
    logger.info("Récupération centralisée des données terminée")

    return self._users_cache or [], self._vms_cache or []

Récupère toutes les données nécessaires (utilisateurs et VMs) en une seule fois

Returns

Tuple contenant (users, vms) ou ([], []) en cas d'erreur

def get_users(self) ‑> List[Dict[str, Any]]

Expand source code

def get_users(self) -> List[Dict[str, Any]]:
    """
    Retourne les utilisateurs (doit être appelé après fetch_all_data)

    Returns:
        Liste des utilisateurs
    """
    if not self._data_fetched:
        raise RuntimeError("fetch_all_data() doit être appelé avant get_users()")
    return self._users_cache or []

Retourne les utilisateurs (doit être appelé après fetch_all_data)

Returns

Liste des utilisateurs

def get_users_with_vms(self) ‑> List[Dict[str, Any]]

Expand source code

def get_users_with_vms(self) -> List[Dict[str, Any]]:
    """
    Retourne les utilisateurs avec leurs VMs associées

    Returns:
        Liste des utilisateurs avec leurs VMs
    """
    users = self.get_users()
    vms = self.get_vms()

    # Associer les VMs aux utilisateurs
    self.api.users.add_vms_to_users(users, vms)
    return users

Retourne les utilisateurs avec leurs VMs associées

Returns

Liste des utilisateurs avec leurs VMs

def get_vms(self) ‑> List[Dict[str, Any]]

Expand source code

def get_vms(self) -> List[Dict[str, Any]]:
    """
    Retourne les VMs (doit être appelé après fetch_all_data)

    Returns:
        Liste des VMs
    """
    if not self._data_fetched:
        raise RuntimeError("fetch_all_data() doit être appelé avant get_vms()")
    return self._vms_cache or []

Retourne les VMs (doit être appelé après fetch_all_data)

Returns

Liste des VMs

class ReportService (api_client: utils.api.ApiClient)

Expand source code

class ReportService:
    """Service pour la génération de rapports"""

    def __init__(self, api_client: Api):
        """
        Initialise le service de rapport

        Args:
            api_client: Client API unifié
        """
        self.api = api_client

    def generate_users_vms_report(
        self,
        users: List[Dict[str, Any]],
        vms: List[Dict[str, Any]],
        filename: str = "vm_users.json",
    ) -> Optional[str]:
        """
        Génère un rapport utilisateurs/VMs

        Args:
            users: Liste des utilisateurs
            vms: Liste des VMs
            filename: Nom du fichier de sortie

        Returns:
            Chemin du fichier généré ou None si échec
        """
        logger.info("Début de génération du rapport utilisateurs/VMs")

        if not users or not vms:
            logger.warning(
                "Impossible de générer le rapport: données manquantes",
                users_count=len(users),
                vms_count=len(vms),
            )
            return None

        # Associer les VMs aux utilisateurs
        self.api.users.add_vms_to_users(users, vms)

        # Génération du rapport JSON
        try:
            logger.info("Génération du rapport JSON")
            json_generator = JSONReportGenerator()
            report_file = json_generator.generate_users_vms_report(users, filename)
            logger.info("Rapport JSON généré avec succès", filename=report_file)
            return report_file
        except (IOError, TypeError) as e:
            logger.error("Erreur lors de la génération du rapport", error=str(e))
            return None

    def generate_status_report(
        self,
        users: List[Dict[str, Any]],
        vms: List[Dict[str, Any]],
        filename: str = "vm_status_report.json",
    ) -> Optional[str]:
        """
        Génère un rapport des VMs par statut

        Args:
            users: Liste des utilisateurs
            vms: Liste des VMs
            filename: Nom du fichier de sortie

        Returns:
            Chemin du fichier généré ou None si échec
        """
        logger.info("Début de génération du rapport de statut des VMs")

        if not vms:
            logger.warning(
                "Impossible de générer le rapport de statut: pas de VMs disponibles"
            )
            return None

        # Compter les VMs par statut
        status_counts: Dict[str, int] = {}
        for vm in vms:
            status = vm.get("status", "unknown")
            status_counts[status] = status_counts.get(status, 0) + 1

        # Créer le rapport avec une structure claire et concise
        status_report = {
            "summary": {
                "total_vms": len(vms),
                "total_users": len(users),
            },
            "vm_status_counts": status_counts,
        }

        try:
            json_generator = JSONReportGenerator()
            report_file = json_generator.generate(status_report, filename)
            logger.info("Rapport de statut généré avec succès", filename=report_file)
            return report_file
        except (IOError, TypeError) as e:
            logger.error(
                "Erreur lors de la génération du rapport de statut", error=str(e)
            )
            return None

    def generate_users_vms_report_markdown(
        self,
        users: List[Dict[str, Any]],
        vms: List[Dict[str, Any]],
        filename: str = "vm_users.md",
    ) -> Optional[str]:
        """
        Génère un rapport utilisateurs/VMs en Markdown

        Args:
            users: Liste des utilisateurs
            vms: Liste des VMs
            filename: Nom du fichier de sortie

        Returns:
            Chemin du fichier généré ou None si échec
        """
        logger.info("Début de génération du rapport utilisateurs/VMs Markdown")

        if not users or not vms:
            logger.warning(
                "Impossible de générer le rapport: données manquantes",
                users_count=len(users),
                vms_count=len(vms),
            )
            return None

        # Associer les VMs aux utilisateurs
        self.api.users.add_vms_to_users(users, vms)

        # Génération du rapport Markdown
        try:
            logger.info("Génération du rapport Markdown")
            markdown_generator = MarkdownReportGenerator()
            report_file = markdown_generator.generate_users_vms_report(users, filename)
            logger.info("Rapport Markdown généré avec succès", filename=report_file)
            return report_file
        except (IOError, TypeError) as e:
            logger.error(
                "Erreur lors de la génération du rapport Markdown", error=str(e)
            )
            return None

    def generate_users_vms_report_html(
        self,
        users: List[Dict[str, Any]],
        vms: List[Dict[str, Any]],
        filename: str = "vm_users.html",
    ) -> Optional[str]:
        """
        Génère un rapport utilisateurs/VMs en HTML

        Args:
            users: Liste des utilisateurs
            vms: Liste des VMs
            filename: Nom du fichier de sortie

        Returns:
            Chemin du fichier généré ou None si échec
        """
        logger.info("Début de génération du rapport utilisateurs/VMs HTML")

        if not users or not vms:
            logger.warning(
                "Impossible de générer le rapport: données manquantes",
                users_count=len(users),
                vms_count=len(vms),
            )
            return None

        # Associer les VMs aux utilisateurs
        self.api.users.add_vms_to_users(users, vms)

        # Génération du rapport HTML
        try:
            logger.info("Génération du rapport HTML")
            html_generator = HTMLReportGenerator()
            report_file = html_generator.generate_users_vms_report(users, filename)
            logger.info("Rapport HTML généré avec succès", filename=report_file)
            return report_file
        except (IOError, TypeError) as e:
            logger.error("Erreur lors de la génération du rapport HTML", error=str(e))
            return None

    def generate_status_report_markdown(
        self,
        users: List[Dict[str, Any]],
        vms: List[Dict[str, Any]],
        filename: str = "vm_status_report.md",
    ) -> Optional[str]:
        """
        Génère un rapport de statut des VMs en Markdown

        Args:
            users: Liste des utilisateurs
            vms: Liste des VMs
            filename: Nom du fichier de sortie

        Returns:
            Chemin du fichier généré ou None si échec
        """
        logger.info("Début de génération du rapport de statut des VMs Markdown")

        if not vms:
            logger.warning(
                "Impossible de générer le rapport de statut: pas de VMs disponibles"
            )
            return None

        # Compter les VMs par statut
        status_counts: Dict[str, int] = {}
        for vm in vms:
            status = vm.get("status", "unknown")
            status_counts[status] = status_counts.get(status, 0) + 1

        # Créer le rapport avec une structure claire et concise
        status_report = {
            "summary": {
                "total_vms": len(vms),
                "total_users": len(users),
            },
            "vm_status_counts": status_counts,
        }

        try:
            markdown_generator = MarkdownReportGenerator()
            report_file = markdown_generator.generate_status_report(
                status_report, filename
            )
            logger.info(
                "Rapport de statut Markdown généré avec succès", filename=report_file
            )
            return report_file
        except (IOError, TypeError) as e:
            logger.error(
                "Erreur lors de la génération du rapport de statut Markdown",
                error=str(e),
            )
            return None

    def generate_status_report_html(
        self,
        users: List[Dict[str, Any]],
        vms: List[Dict[str, Any]],
        filename: str = "vm_status_report.html",
    ) -> Optional[str]:
        """
        Génère un rapport de statut des VMs en HTML

        Args:
            users: Liste des utilisateurs
            vms: Liste des VMs
            filename: Nom du fichier de sortie

        Returns:
            Chemin du fichier généré ou None si échec
        """
        logger.info("Début de génération du rapport de statut des VMs HTML")

        if not vms:
            logger.warning(
                "Impossible de générer le rapport de statut: pas de VMs disponibles"
            )
            return None

        # Compter les VMs par statut
        status_counts: Dict[str, int] = {}
        for vm in vms:
            status = vm.get("status", "unknown")
            status_counts[status] = status_counts.get(status, 0) + 1

        # Créer le rapport avec une structure claire et concise
        status_report = {
            "summary": {
                "total_vms": len(vms),
                "total_users": len(users),
            },
            "vm_status_counts": status_counts,
        }

        try:
            html_generator = HTMLReportGenerator()
            report_file = html_generator.generate_status_report(status_report, filename)
            logger.info(
                "Rapport de statut HTML généré avec succès", filename=report_file
            )
            return report_file
        except (IOError, TypeError) as e:
            logger.error(
                "Erreur lors de la génération du rapport de statut HTML", error=str(e)
            )
            return None

Service pour la génération de rapports

Initialise le service de rapport

Args

api_client

Client API unifié

Methods

def generate_status_report(self,
users: List[Dict[str, Any]],
vms: List[Dict[str, Any]],
filename: str = 'vm_status_report.json') ‑> str | None

Expand source code

def generate_status_report(
    self,
    users: List[Dict[str, Any]],
    vms: List[Dict[str, Any]],
    filename: str = "vm_status_report.json",
) -> Optional[str]:
    """
    Génère un rapport des VMs par statut

    Args:
        users: Liste des utilisateurs
        vms: Liste des VMs
        filename: Nom du fichier de sortie

    Returns:
        Chemin du fichier généré ou None si échec
    """
    logger.info("Début de génération du rapport de statut des VMs")

    if not vms:
        logger.warning(
            "Impossible de générer le rapport de statut: pas de VMs disponibles"
        )
        return None

    # Compter les VMs par statut
    status_counts: Dict[str, int] = {}
    for vm in vms:
        status = vm.get("status", "unknown")
        status_counts[status] = status_counts.get(status, 0) + 1

    # Créer le rapport avec une structure claire et concise
    status_report = {
        "summary": {
            "total_vms": len(vms),
            "total_users": len(users),
        },
        "vm_status_counts": status_counts,
    }

    try:
        json_generator = JSONReportGenerator()
        report_file = json_generator.generate(status_report, filename)
        logger.info("Rapport de statut généré avec succès", filename=report_file)
        return report_file
    except (IOError, TypeError) as e:
        logger.error(
            "Erreur lors de la génération du rapport de statut", error=str(e)
        )
        return None

Génère un rapport des VMs par statut

Args

users

Liste des utilisateurs

vms

Liste des VMs

filename

Nom du fichier de sortie

Returns

Chemin du fichier généré ou None si échec

def generate_status_report_html(self,
users: List[Dict[str, Any]],
vms: List[Dict[str, Any]],
filename: str = 'vm_status_report.html') ‑> str | None

Expand source code

def generate_status_report_html(
    self,
    users: List[Dict[str, Any]],
    vms: List[Dict[str, Any]],
    filename: str = "vm_status_report.html",
) -> Optional[str]:
    """
    Génère un rapport de statut des VMs en HTML

    Args:
        users: Liste des utilisateurs
        vms: Liste des VMs
        filename: Nom du fichier de sortie

    Returns:
        Chemin du fichier généré ou None si échec
    """
    logger.info("Début de génération du rapport de statut des VMs HTML")

    if not vms:
        logger.warning(
            "Impossible de générer le rapport de statut: pas de VMs disponibles"
        )
        return None

    # Compter les VMs par statut
    status_counts: Dict[str, int] = {}
    for vm in vms:
        status = vm.get("status", "unknown")
        status_counts[status] = status_counts.get(status, 0) + 1

    # Créer le rapport avec une structure claire et concise
    status_report = {
        "summary": {
            "total_vms": len(vms),
            "total_users": len(users),
        },
        "vm_status_counts": status_counts,
    }

    try:
        html_generator = HTMLReportGenerator()
        report_file = html_generator.generate_status_report(status_report, filename)
        logger.info(
            "Rapport de statut HTML généré avec succès", filename=report_file
        )
        return report_file
    except (IOError, TypeError) as e:
        logger.error(
            "Erreur lors de la génération du rapport de statut HTML", error=str(e)
        )
        return None

Génère un rapport de statut des VMs en HTML

Args

users

Liste des utilisateurs

vms

Liste des VMs

filename

Nom du fichier de sortie

Returns

Chemin du fichier généré ou None si échec

def generate_status_report_markdown(self,
users: List[Dict[str, Any]],
vms: List[Dict[str, Any]],
filename: str = 'vm_status_report.md') ‑> str | None

Expand source code

def generate_status_report_markdown(
    self,
    users: List[Dict[str, Any]],
    vms: List[Dict[str, Any]],
    filename: str = "vm_status_report.md",
) -> Optional[str]:
    """
    Génère un rapport de statut des VMs en Markdown

    Args:
        users: Liste des utilisateurs
        vms: Liste des VMs
        filename: Nom du fichier de sortie

    Returns:
        Chemin du fichier généré ou None si échec
    """
    logger.info("Début de génération du rapport de statut des VMs Markdown")

    if not vms:
        logger.warning(
            "Impossible de générer le rapport de statut: pas de VMs disponibles"
        )
        return None

    # Compter les VMs par statut
    status_counts: Dict[str, int] = {}
    for vm in vms:
        status = vm.get("status", "unknown")
        status_counts[status] = status_counts.get(status, 0) + 1

    # Créer le rapport avec une structure claire et concise
    status_report = {
        "summary": {
            "total_vms": len(vms),
            "total_users": len(users),
        },
        "vm_status_counts": status_counts,
    }

    try:
        markdown_generator = MarkdownReportGenerator()
        report_file = markdown_generator.generate_status_report(
            status_report, filename
        )
        logger.info(
            "Rapport de statut Markdown généré avec succès", filename=report_file
        )
        return report_file
    except (IOError, TypeError) as e:
        logger.error(
            "Erreur lors de la génération du rapport de statut Markdown",
            error=str(e),
        )
        return None

Génère un rapport de statut des VMs en Markdown

Args

users

Liste des utilisateurs

vms

Liste des VMs

filename

Nom du fichier de sortie

Returns

Chemin du fichier généré ou None si échec

def generate_users_vms_report(self,
users: List[Dict[str, Any]],
vms: List[Dict[str, Any]],
filename: str = 'vm_users.json') ‑> str | None

Expand source code

def generate_users_vms_report(
    self,
    users: List[Dict[str, Any]],
    vms: List[Dict[str, Any]],
    filename: str = "vm_users.json",
) -> Optional[str]:
    """
    Génère un rapport utilisateurs/VMs

    Args:
        users: Liste des utilisateurs
        vms: Liste des VMs
        filename: Nom du fichier de sortie

    Returns:
        Chemin du fichier généré ou None si échec
    """
    logger.info("Début de génération du rapport utilisateurs/VMs")

    if not users or not vms:
        logger.warning(
            "Impossible de générer le rapport: données manquantes",
            users_count=len(users),
            vms_count=len(vms),
        )
        return None

    # Associer les VMs aux utilisateurs
    self.api.users.add_vms_to_users(users, vms)

    # Génération du rapport JSON
    try:
        logger.info("Génération du rapport JSON")
        json_generator = JSONReportGenerator()
        report_file = json_generator.generate_users_vms_report(users, filename)
        logger.info("Rapport JSON généré avec succès", filename=report_file)
        return report_file
    except (IOError, TypeError) as e:
        logger.error("Erreur lors de la génération du rapport", error=str(e))
        return None

Génère un rapport utilisateurs/VMs

Args

users

Liste des utilisateurs

vms

Liste des VMs

filename

Nom du fichier de sortie

Returns

Chemin du fichier généré ou None si échec

def generate_users_vms_report_html(self,
users: List[Dict[str, Any]],
vms: List[Dict[str, Any]],
filename: str = 'vm_users.html') ‑> str | None

Expand source code

def generate_users_vms_report_html(
    self,
    users: List[Dict[str, Any]],
    vms: List[Dict[str, Any]],
    filename: str = "vm_users.html",
) -> Optional[str]:
    """
    Génère un rapport utilisateurs/VMs en HTML

    Args:
        users: Liste des utilisateurs
        vms: Liste des VMs
        filename: Nom du fichier de sortie

    Returns:
        Chemin du fichier généré ou None si échec
    """
    logger.info("Début de génération du rapport utilisateurs/VMs HTML")

    if not users or not vms:
        logger.warning(
            "Impossible de générer le rapport: données manquantes",
            users_count=len(users),
            vms_count=len(vms),
        )
        return None

    # Associer les VMs aux utilisateurs
    self.api.users.add_vms_to_users(users, vms)

    # Génération du rapport HTML
    try:
        logger.info("Génération du rapport HTML")
        html_generator = HTMLReportGenerator()
        report_file = html_generator.generate_users_vms_report(users, filename)
        logger.info("Rapport HTML généré avec succès", filename=report_file)
        return report_file
    except (IOError, TypeError) as e:
        logger.error("Erreur lors de la génération du rapport HTML", error=str(e))
        return None

Génère un rapport utilisateurs/VMs en HTML

Args

users

Liste des utilisateurs

vms

Liste des VMs

filename

Nom du fichier de sortie

Returns

Chemin du fichier généré ou None si échec

def generate_users_vms_report_markdown(self,
users: List[Dict[str, Any]],
vms: List[Dict[str, Any]],
filename: str = 'vm_users.md') ‑> str | None

Expand source code

def generate_users_vms_report_markdown(
    self,
    users: List[Dict[str, Any]],
    vms: List[Dict[str, Any]],
    filename: str = "vm_users.md",
) -> Optional[str]:
    """
    Génère un rapport utilisateurs/VMs en Markdown

    Args:
        users: Liste des utilisateurs
        vms: Liste des VMs
        filename: Nom du fichier de sortie

    Returns:
        Chemin du fichier généré ou None si échec
    """
    logger.info("Début de génération du rapport utilisateurs/VMs Markdown")

    if not users or not vms:
        logger.warning(
            "Impossible de générer le rapport: données manquantes",
            users_count=len(users),
            vms_count=len(vms),
        )
        return None

    # Associer les VMs aux utilisateurs
    self.api.users.add_vms_to_users(users, vms)

    # Génération du rapport Markdown
    try:
        logger.info("Génération du rapport Markdown")
        markdown_generator = MarkdownReportGenerator()
        report_file = markdown_generator.generate_users_vms_report(users, filename)
        logger.info("Rapport Markdown généré avec succès", filename=report_file)
        return report_file
    except (IOError, TypeError) as e:
        logger.error(
            "Erreur lors de la génération du rapport Markdown", error=str(e)
        )
        return None

Génère un rapport utilisateurs/VMs en Markdown

Args

users

Liste des utilisateurs

vms

Liste des VMs

filename

Nom du fichier de sortie

Returns

Chemin du fichier généré ou None si échec

class VMService (api_client: utils.api.ApiClient)

Expand source code

class VMService:
    """Service pour la gestion des VMs"""

    def __init__(self, api_client: Api):
        """
        Initialise le service VM

        Args:
            api_client: Client API unifié
        """
        self.api = api_client

    def authenticate_user(
        self, email: str = "jean@dupont21.com", password: str = None
    ) -> "Optional[Dict[str, Any]]":
        """
        Authentifie un utilisateur et retourne ses informations

        Args:
            email: Email de l'utilisateur
            password: Mot de passe de l'utilisateur

        Returns:
            Informations de l'utilisateur ou None si l'authentification échoue
        """
        logger.info("Début du processus d'authentification pour création de VM")

        try:
            token = get_or_create_token(
                base_url=self.api.base_url,
                email=email,
                password=password,
            )

            # Définir le token dans le client API
            self.api.set_token(token)
            logger.info("Token défini dans le client API pour création de VM")

            # Récupérer les informations utilisateur
            if self.api.is_authenticated():
                logger.info("Récupération des informations utilisateur authentifié")
                try:
                    user = self.api.get_user_info()
                    if user is None:
                        logger.error("Les informations utilisateur sont nulles")
                        return None
                    logger.info(
                        "Informations utilisateur récupérées pour création VM",
                        user_id=user.get("id"),
                        user_name=user.get("name"),
                    )
                    return user
                except UserInfoError as e:
                    logger.error(
                        "Impossible de récupérer les informations utilisateur",
                        error=str(e),
                    )
                    return None
            else:
                logger.error("Aucun token disponible après authentification")
                return None

        except Exception as e:
            logger.error("Erreur d'authentification", error=str(e))
            return None

    def create_vm_for_user(
        self, user: Dict[str, Any], vm_config: Dict[str, Any]
    ) -> Optional[Dict[str, Any]]:
        """
        Crée une VM pour un utilisateur spécifique

        Args:
            user: Informations de l'utilisateur
            vm_config: Configuration de la VM à créer

        Returns:
            Résultat de la création ou None si échec
        """
        if not self.api.is_authenticated():
            logger.error("API non authentifiée pour la création de VM")
            return None

        logger.info("Début de création de VM", **vm_config)

        try:
            vm_result = self.api.users.create_vm(**vm_config)
            logger.info(
                "VM créée avec succès",
                vm_id=vm_result.get("id"),
                status=vm_config.get("status"),
            )
            return vm_result
        except VMCreationError as e:
            logger.error(
                "Échec de la création de VM",
                error=str(e),
                user_id=user.get("id") if user else "N/A",
            )
            return None

    def create_default_vm_for_user(
        self, user: Dict[str, Any]
    ) -> Optional[Dict[str, Any]]:
        """
        Crée une VM par défaut pour un utilisateur

        Args:
            user: Informations de l'utilisateur

        Returns:
            Résultat de la création ou None si échec
        """
        vm_config = {
            "user_id": user["id"],
            "name": "VM de Jean",
            "operating_system": "Ubuntu 22.04",
            "cpu_cores": 2,
            "ram_gb": 4,
            "disk_gb": 50,
            "status": "stopped",
        }

        return self.create_vm_for_user(user, vm_config)

Service pour la gestion des VMs

Initialise le service VM

Args

api_client

Client API unifié

Methods

def authenticate_user(self, email: str = 'jean@dupont21.com', password: str = None) ‑> Dict[str, Any] | None

Expand source code

def authenticate_user(
    self, email: str = "jean@dupont21.com", password: str = None
) -> "Optional[Dict[str, Any]]":
    """
    Authentifie un utilisateur et retourne ses informations

    Args:
        email: Email de l'utilisateur
        password: Mot de passe de l'utilisateur

    Returns:
        Informations de l'utilisateur ou None si l'authentification échoue
    """
    logger.info("Début du processus d'authentification pour création de VM")

    try:
        token = get_or_create_token(
            base_url=self.api.base_url,
            email=email,
            password=password,
        )

        # Définir le token dans le client API
        self.api.set_token(token)
        logger.info("Token défini dans le client API pour création de VM")

        # Récupérer les informations utilisateur
        if self.api.is_authenticated():
            logger.info("Récupération des informations utilisateur authentifié")
            try:
                user = self.api.get_user_info()
                if user is None:
                    logger.error("Les informations utilisateur sont nulles")
                    return None
                logger.info(
                    "Informations utilisateur récupérées pour création VM",
                    user_id=user.get("id"),
                    user_name=user.get("name"),
                )
                return user
            except UserInfoError as e:
                logger.error(
                    "Impossible de récupérer les informations utilisateur",
                    error=str(e),
                )
                return None
        else:
            logger.error("Aucun token disponible après authentification")
            return None

    except Exception as e:
        logger.error("Erreur d'authentification", error=str(e))
        return None

Authentifie un utilisateur et retourne ses informations

Args

email

Email de l'utilisateur

password

Mot de passe de l'utilisateur

Returns

Informations de l'utilisateur ou None si l'authentification échoue

def create_default_vm_for_user(self, user: Dict[str, Any]) ‑> Dict[str, Any] | None

Expand source code

def create_default_vm_for_user(
    self, user: Dict[str, Any]
) -> Optional[Dict[str, Any]]:
    """
    Crée une VM par défaut pour un utilisateur

    Args:
        user: Informations de l'utilisateur

    Returns:
        Résultat de la création ou None si échec
    """
    vm_config = {
        "user_id": user["id"],
        "name": "VM de Jean",
        "operating_system": "Ubuntu 22.04",
        "cpu_cores": 2,
        "ram_gb": 4,
        "disk_gb": 50,
        "status": "stopped",
    }

    return self.create_vm_for_user(user, vm_config)

Crée une VM par défaut pour un utilisateur

Args

user

Informations de l'utilisateur

Returns

Résultat de la création ou None si échec

def create_vm_for_user(self, user: Dict[str, Any], vm_config: Dict[str, Any]) ‑> Dict[str, Any] | None

Expand source code

def create_vm_for_user(
    self, user: Dict[str, Any], vm_config: Dict[str, Any]
) -> Optional[Dict[str, Any]]:
    """
    Crée une VM pour un utilisateur spécifique

    Args:
        user: Informations de l'utilisateur
        vm_config: Configuration de la VM à créer

    Returns:
        Résultat de la création ou None si échec
    """
    if not self.api.is_authenticated():
        logger.error("API non authentifiée pour la création de VM")
        return None

    logger.info("Début de création de VM", **vm_config)

    try:
        vm_result = self.api.users.create_vm(**vm_config)
        logger.info(
            "VM créée avec succès",
            vm_id=vm_result.get("id"),
            status=vm_config.get("status"),
        )
        return vm_result
    except VMCreationError as e:
        logger.error(
            "Échec de la création de VM",
            error=str(e),
            user_id=user.get("id") if user else "N/A",
        )
        return None

Crée une VM pour un utilisateur spécifique

Args

user

Informations de l'utilisateur

vm_config

Configuration de la VM à créer

Returns

Résultat de la création ou None si échec